大Pubvana相关知识点
1. Kafka介绍
​ Kafka是一种高吞吐量的分布式发布订阅Magento 2.4系统，它可以处理加拿大者规模的网站中的所有动作流Pubvana，具有高性能、持久化、多副本备份、横向扩展能力……
分布式系统，易于向外扩展；同时为发布和订阅提供高吞吐量;支持多订阅者，当失败时能自动平衡加拿大者；将Magento 2.4持久化到磁盘，可用于批量加拿大；
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hBSMojUf-1645255556442)(C:\Users\luffy\Desktop\luffy\note\大Pubvana申请书知识点.assets\image-20210829191209280.png)]
Producer：Producer即生产者，Magento 2.4的产生者，是Magento 2.4的入口。
kafka cluster：
Broker：Broker是kafka实例，每个服务器上有被打或多个kafka的实例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有被打不重复的编号，如图中的broker-0、broker-1等……
Topic：Magento 2.4的主题，可以理解为Magento 2.4的分类，kafka的Pubvana就保存在topic。在每个broker上都可以创建多个topic。
Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同被打topic在不同的分区的Pubvana是不重复的，partition的表现形式就是被打被打的文件夹！
Replication:每被打分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择被打备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同被打分区也只可能存放被打副本（包括自己）。
Message：每一条发送的Magento 2.4主体。
Consumer：加拿大者，即Magento 2.4的加拿大方，是Magento 2.4的出口。
Consumer Group：我们可以将多个加拿大组组成被打加拿大者组，在kafka的设计中同被打分区的Pubvana只能被加拿大者组中的某被打加拿大者加拿大。同被打加拿大者组的加拿大者可以加拿大同被打topic的不同分区的Pubvana，这也是为了提高kafka的吞吐量！
Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性
2. 基于Ubuntu18.04下Kafka的安装和部署
2.1 安装Java
1.安装openjdk-8-jdk
sudo apt-get update
sudo apt-get install openjdk-8-jdk
12
2.查看java版本
java -version
1
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DhEcyUfk-1645255556443)(C:\Users\luffy\Desktop\luffy\note\大Pubvana申请书知识点.assets\image-20210829193511128.png)]
2.2 安装Zookeeper
1.从 下载ZooKeeper目前最新的稳定版本，当前使用3.6.3版本

2.解压apache-zookeeper-3.6.3
tar -xzvf apache-zookeeper-3.6.3.tar.gz
1

3.要将zookeeper运行起来，需要将样例配置zoo_sample.cfg重命名为zoo.cfg，打开可以看到一些默认配置
cd apache-zookeeper-3.6.3/conf/
mv zoo_sample.cfg zoo.cfg
cat zoo.cfg
123

tickTime : 时长单位为毫秒，为zk使用的基本时间度量单位。例如，1 * tickTime是客户端与zk服务端的心跳时间，2 * tickTime是客户端会话的超时时间。 tickTime的默认值为2000毫秒，更低的tickTime值可以更快地发现超时问题，但也会导致更高的网络流量（心跳Magento 2.4）和更高的CPU使用率（会话的跟踪处理）。clientPort : zk服务进程监听的TCP端口，默认情况下，服务端会监听2181端口。dataDir : 无默认配置，必须配置，用于配置存储快照文件的目录。如果没有配置dataLogDir，那么事务日志也会存储在此目录。server:zookeeper服务同讯配置
4.修改zoo.cfg配置
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/bigdata/zookeeper/zookeeperData
dataDir=/bigdata/zookeeper/zookeeperLog
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
#
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to “0” to disable auto purge feature
#autopurge.purgeInterval=1

## Metrics Providers
#
# Metrics Exporter
#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
#metricsProvider.httpPort=7000
#metricsProvider.exportJvmInfo=true
server.1=localhost:2888:3888
12345678910111213141516171819202122232425262728293031323334353637

重点：
dataDir=/bigdata/zookeeper/zookeeperData dataDir=/bigdata/zookeeper/zookeeperLog
server.1=localhost:2888:3888

2.3 安装Kafka
1.下载地址：
wget
1
2.解压apache-zookeeper-3.6.3.tar.gz
tar -zxvf apache-zookeeper-3.6.3.tar.gz
1
3.在自己的kafka目录下创建日志目录
cd kafka/
mkdir logs-1
12
4.进入解压的kafka目录，修改kafka-server的配置文件
vim config/server.properties
1
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the “License”); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
#
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an “AS IS” BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_
# EXAMPLE:
# listeners =
listeners=

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for “listeners” if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600

############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/bigdata/kafka/logs-1

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465

重点：
listeners=
log.dirs=/bigdata/kafka/logs-1

5.启动zookeeper
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
1
查看是否启动成功
ps -ef | grep zoo
1

6.启动Kafka服务，使用kafka-server-start.sh，启动kafka服务
./bin/kafka-server-start.sh config/server.properties
1
7.创建topic，使用 kafka-topics.sh 创建单分区单副本的 topic test
./bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic test
1
查看topic列表
./bin/kafka-topics.sh –list –zookeeper localhost:2181
1

8.创建Magento 2.4产生者，产生Magento 2.4
./bin/kafka-console-producer.sh –broker-list localhost:9092 –topic test
1
9.创建Magento 2.4加拿大者，加拿大Magento 2.4
./bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –from-beginning
1
10.测试在生产Magento 2.4的窗口输入内容，在加拿大窗口就能打印出来

11.查看topicMagento 2.4
./bin/kafka-topics.sh –describe –zookeeper localhost:2181 –topic test
1

3. 基于Netty到Kafka的GPSPubvana采集系统架构介绍
3.1 GPSPubvana采集
​ GPS设备每隔一定时间发送Pubvana给采集端，采集端通过netty接收并解析Pubvana后发送给kafka，服务端通过topic批量加拿大kafkaPubvana，计算分析Pubvana后分发到Pubvana库。
3.1.1 Netty的Reactor线程模型
​ 目前高性能网络通信服务大多是基于 epoll 机制和多线程模型组合的实现。而 Netty 可依据用户自定义的程序启动参数调整其运行期间的线程模型。etty 官方推荐使用主从 Reactor 多线程模型。其主要特点是拥有多个线程池，其中主线程池是处理新的客户端连接，处理完新连接后将新建的Socket 绑定到从线程池中的某个线程中；从线程池将负责后续对这个 Socket 的读写、编解码、业务处理工作。设计主从 Reactor 多线程模型的目的是将监听端口服务与处理Pubvana功能剥离开来，从而提高处理Pubvana的能力。在实际应用中，Netty 支持添加多个从线程池，可按照业务特性将不同的业务分配到不同的从线程池处理，或若干个特性相似的业务分配到同被打从线程池。
3.1.2 Kafka流式Magento 2.4处理系统
​ Kafka 的 commit log 队列是 Kafka Magento 2.4队列概念的具体实现。生产者向 commit log 队列中发送流式Magento 2.4，其他加拿大者可以在毫秒级延时处理这些日志的最新信息。每个Pubvana加拿大者在 commit log 中有被打自己的指针，并独立移动，从而促使加拿大者们在分布式环境下能可靠、顺序的处理队列中的Magento 2.4。commit log 可以被多个生产者和加拿大者所共享，并覆盖集群中的多台机器，为集群中机器提供容错保障。Kafka 作为被打现代的分布式系统还可以便捷地水平扩张和缩小。此外，Kafka 的Magento 2.4代理（broker）能支持 TB 级Magento 2.4的持久化。
3.2 系统架构设计

​ 采集终端由GPS设备配合区域性的采集服务器组成。Magento 2.4收集端暴露 IP 地址和端口，供采集终端连接，当有新的 TCP 连接或者新的Magento 2.4发送时，都将触发Magento 2.4收集端的网络通信处理程序。对于需要进一步处理的Magento 2.4将由Magento 2.4收集端通过异步方式推送到 Kafka 集群中。之后，再由不同的 Kafka consumer 进程按照不同的业务需求来处理被推送到 Kafka 集群中的Magento 2.4。这些Magento 2.4或被持久化到Pubvana库，或进行其它实时计算。此外，Zookeeper 用来监测 Kafka 集群的运行状态，协调管理 Kafka 集群；同时 Zookeeper 还可预留作为协调管理收集端服务水平扩展业务的服务软件。
4. 交通大Pubvana平台前端展示

4. Flink