千云物流 -LBS[GPS]服务:kafka集群部署

需要环境和机器【10.0.0.7】

准备3台CentOS服务器，并配置好静态IP、主机名

//对应的配置样例。
BOOTPROTO=static
IPADDR=10.0.0.80   # 设置的静态IP地址
NETMASK=255.255.255.0    # 子网掩码
GATEWAY=10.0.0.1    # 网关地址
DNS1=10.0.0.1       # DNS服务器

软件版本说明

ZooKeeper集群
Kakfa集群需要依赖ZooKeeper存储Broker、Topic等信息，这里我们部署三台ZK，对应的和kafka同样的机器。

按装部署ZK集群

下载ZooKeeper&基础准备
官方镜像选择：https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/

cd /home/download
wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.8.0/apache-zookeeper-3.8.0-bin.tar.gz
wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.8.0/apache-zookeeper-3.8.0.tar.gz

创建ZooKeeper相关目录

#创建应用目录
mkdir /usr/zookeeper

#创建数据目录
mkdir /zookeeper
mkdir /zookeeper/data
mkdir /zookeeper/logs

解压到指定目录

tar -zvxf apache-zookeeper-3.8.0-bin.tar.gz -C /usr/zookeeper

配置环境变量

#修改环境变量文件
vi /etc/profile

#增加以下内容
export ZOOKEEPER_HOME=/usr/zookeeper/apache-zookeeper-3.8.0
export PATH=$ZOOKEEPER_HOME/bin:$PATH

#使环境变量生效
source /etc/profile

#查看配置结果
echo $ZOOKEEPER_HOME

既然已配置环境变量，为了方便访问ZooKeeper目录
后续通过$ZOOKEEPER_HOME代替/usr/zookeeper/apcche-zookeeper-3.8.0

配置ZooKeeper
#进入ZooKeeper配置目录

cd $ZOOKEEPER_HOME/conf
#新建配置文件
vi zoo.cfg
#写入以下内容并保存
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/zookeeper/data
dataLogDir=/zookeeper/logs
clientPort=2181
server.1=10.0.0.80:2888:3888
server.2=10.0.0.83:2888:3888
server.3=10.0.0.85:2888:3888

ZooKeeper常用配置项说明
tickTime	CS通信心跳间隔	服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每间隔 tickTime 时间就会发送一个心跳。tickTime以毫秒为单位。
initLimit	LF初始通信时限	集群中的follower服务器(F)与leader服务器(L)之间初始连接时能容忍的最多心跳数
syncLimit	LF同步通信时限	集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数
dataDir	数据文件目录	Zookeeper保存数据的目录，默认情况下，Zookeeper将写数据的日志文件也保存在这个目录里
dataLogDir	日志文件目录	Zookeeper保存日志文件的目录
clientPort	客户端连接端口	客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求
server.N	服务器名称与地址	从N开始依次为：服务编号、服务地址、LF通信端口、选举端口；例如：server.1=10.0.0.80:2888:3888

配置节点标识
zk01：

echo "1" > /zookeeper/data/myid

zk02：

echo "2" > /zookeeper/data/myid

zk03：

echo "3" > /zookeeper/data/myid

防火墙配置

#开放端口
firewall-cmd --add-port=2181/tcp --permanent
firewall-cmd --add-port=2888/tcp --permanent
firewall-cmd --add-port=3888/tcp --permanent

#重新加载防火墙配置
firewall-cmd --reload

	centOS7 查看防火墙状态
	 一、防火墙的开启、关闭、禁用命令
	
	（1）设置开机启用防火墙：systemctl enable firewalld.service
	
	（2）设置开机禁用防火墙：systemctl disable firewalld.service
	
	（3）启动防火墙：systemctl start firewalld
	
	（4）关闭防火墙：systemctl stop firewalld
	
	（5）检查防火墙状态：systemctl status firewalld 
	
	二、使用firewall-cmd配置端口
	
	（1）查看防火墙状态：firewall-cmd --state
	
	（2）重新加载配置：firewall-cmd --reload
	
	（3）查看开放的端口：firewall-cmd --list-ports
	
	（4）开启防火墙端口：firewall-cmd --zone=public --add-port=9200/tcp --permanent
	
	　　命令含义：
	
	　　–zone #作用域
	
	　　–add-port=9200/tcp #添加端口，格式为：端口/通讯协议
	
	　　–permanent #永久生效，没有此参数重启后失效
	
	　　注意：添加端口后，必须用命令firewall-cmd --reload重新加载一遍才会生效
	
	（5）
	 	添加对外开放端口：firewall-cmd --zone=public --add-port=9092/tcp --permanent
		关闭防火墙端口：firewall-cmd --zone=public --remove-port=9200/tcp --permanent

启动ZooKeeper

#进入ZooKeeper bin目录
cd $ZOOKEEPER_HOME/bin

#启动
sh zkServer.sh start

集群查看&连接测试

查看节点状态
配置互信https://www.cnblogs.com/bjx2020/p/8565880.html

sh $ZOOKEEPER_HOME/bin/zkServer.sh status
# 执行前确定将三台机器进行了ssh互信。
# 错误状态信息
Client address: localhost. Client SSL: false.
Error contacting service. It is probably not running.
1. 检查防火墙端口是否开启
2. 检查配置文件是否配置正确，是否有多余的空格之类的
3. 检查下载的是编译后的zk包还是bin的安装包。
4. 检查你的文件是否载复制时候少了字母，一般会载开头和结尾查找
 
#启动成功状态信息
ZooKeeper JMX enabled by default
Using config: /usr/zookeeper/zookeeper-3.4.11/bin/../conf/zoo.cfg
Mode: follower

#如果为领导者节点则Mode:leader

客户端连接测试
这里随机选其中一个节点作为客户端连接其他节点即可

#指定Server进行连接
sh $ZOOKEEPER_HOME/bin/zkCli.sh -server 10.0.0.80:2181

#正常连接后会进入ZooKeeper命令行，显示如下：
[zk: 10.0.0.80:2181(CONNECTED) 0]

输入命令测试：

#查看ZooKeeper根
[zk: 10.0.0.80:2181(CONNECTED) 0] ls /
[zookeeper]

安装kafka集群

Kafka简介

Kafka是一个开源的分布式消息引擎/消息中间件，同时Kafka也是一个流处理平台。Kakfa支持以发布/订阅的方式在应用间传递消息，同时并基于消息功能添加了Kafka Connect、Kafka Streams以支持连接其他系统的数据(Elasticsearch、Hadoop等)

Kafka最核心的最成熟的还是他的消息引擎，所以Kafka大部分应用场景还是用来作为消息队列削峰平谷。另外，Kafka也是目前性能最好的消息中间件。

Kafka架构

在Kafka集群(Cluster)中，一个Kafka节点就是一个Broker，消息由Topic来承载，可以存储在1个或多个Partition中。发布消息的应用为Producer、消费消息的应用为Consumer，多个Consumer可以促成Consumer Group共同消费一个Topic中的消息。这个和别的消息服务一样。

部署过程

创建目录

mkdir /usr/kafka

#创建Kafka数据目录
mkdir /kafka
mkdir /kafka/logs
chmod 777 -R /kafka

下载&解压
Kafka官方下载地址：https://dlcdn.apache.org/kafka/3.1.0/kafka_2.12-3.1.0.tgz

#创建并进入下载目录
mkdir /home/downloads
cd /home/downloads

#下载安装包
wget https://dlcdn.apache.org/kafka/3.1.0/kafka_2.12-3.1.0.tgz
#解压到应用目录
tar -zvxf kafka_2.12-3.1.0.tgz -C /usr/kafka

Kafka节点配置

#进入应用目录
cd /usr/kafka/kafka_2.12-3.1.0/

#修改配置文件
vi config/server.properties

通用配置
Kafka常用Broker配置说明：

配置日志目录、指定ZooKeeper服务器

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

# root directory for all kafka znodes.
zookeeper.connect=10.0.0.80:2181,10.0.0.83:2181,10.0.0.85:2181

分节点配置
修改server.properties配置

#kafka01 值唯一
broker.id=1
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.0.0.80:9092

#kafka02 值唯一
broker.id=1
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.0.0.83:9092

#kafka03 值唯一
broker.id=1
#listeners=PLAINTEXT://:9092
listeners=PLAINTEXT://10.0.0.85:9092

防火墙配置

#开放端口
firewall-cmd --add-port=9092/tcp --permanent

#重新加载防火墙配置
firewall-cmd --reload

启动Kafka

#进入kafka根目录
cd /usr/kafka/kafka_2.12-2.3.0/
#启动
# ./bin/kafka-server-start.sh -daemon config/server.properties &
./bin/kafka-server-start.sh config/server.properties &

#启动成功输出示例(最后几行)

[2022-03-14 18:23:23,382] INFO Kafka version: 3.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-14 18:23:23,382] INFO Kafka commitId: 37edeed0777bacb3 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-14 18:23:23,382] INFO Kafka startTimeMs: 1647253403379 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-14 18:23:23,383] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)

Kafka测试

创建Topic
在kafka01(Broker)上创建测试Tpoic：test-janle，这里我们指定了3个副本、1个分区

bin/kafka-topics.sh --create --bootstrap-server 10.0.0.80:9092 --replication-factor 3 --partitions 1 --topic test-janle

Topic在kafka01上创建后也会同步到集群中另外两个Broker：kafka02、kafka03

查看Topic
我们可以通过命令列出指定的Broker

bin/kafka-topics.sh --list --bootstrap-server 10.0.0.83:9092

发送消息
这里我们向Broker(id=0)的Topic=test-janle发送消息

bin/kafka-console-producer.sh --broker-list  10.0.0.80:9092  --topic test-janle

#消息内容
> hello word

消费消息
在Kafka02上消费Broker03的消息

bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.85:9092 --topic test-janle --from-beginning

在Kafka03上消费Broker02的消息

bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.83:9092 --topic test-janle  --from-beginning

然后均能收到消息

 hello word

这是因为这两个消费消息的命令是建立了两个不同的Consumer
如果我们启动Consumer指定Consumer Group Id就可以作为一个消费组协同工，1个消息同时只会被一个Consumer消费到

bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.83:9092 --topic test-ken-io --from-beginning --group testgroup_ken
bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.85:9092 --topic test-ken-io --from-beginning --group testgroup_ken

kafka集群管理工具kafka-manager
为了简化开发者和服务工程师维护Kafka集群的工作，yahoo构建了一个叫做Kafka管理器的基于Web工具，叫做 Kafka Manager。kafka-manager 项目地址：https://github.com/yahoo/kafka-manager。这个管理工具可以很容易地发现分布在集群中的哪些topic分布不均匀，或者是分区在整个集群分布不均匀的的情况。它支持管理多个集群、选择副本、副本重新分配以及创建Topic。这个管理工具也是一个非常好的可以快速浏览这个集群的工具，https://github.com/yahoo/CMAK/releases
kafka-manager有如下功能：

• 管理多个kafka集群
• 便捷的检查kafka集群状态(topics,brokers,备份分布情况,分区分布情况)
• 选择你要运行的副本
• 基于当前分区状况进行
• 可以选择topic配置并创建topic(0.8.1.1和0.8.2的配置不同)
• 删除topic(只支持0.8.2以上的版本并且要在broker配置中设置delete.topic.enable=true)
• Topic list会指明哪些topic被删除（在0.8.2以上版本适用）
• 为已存在的topic增加分区
• 为已存在的topic更新配置
• 在多个topic上批量重分区
• 在多个topic上批量重分区(可选partition broker位置)

部署kafka-manager

编译好的kafka-manager-1.3.3.23.zip，这里建议使用已经编译好的，你懂的。

链接：https://pan.百度.com/s/1HLQjDRIGnKjiwX9V_2jjYg?pwd=iwzc 
提取码：iwzc

#java 1.8的话建议使用，如果是新版本需要java11以上。1.0后的需要编译比较麻烦，使用已经编译好的就可以。
cd kafka-manager-1.3.3.23/
cd conf/
vim application.conf
#把zookeeper都写进去
kafka-manager.zkhosts="10.0.0.80:2181,10.0.0.83:2181,10.0.0.85:2181"

#启动kafka-manager
nohup ./bin/kafka-manager-Dhttp.port=9107   > /dev/null 2>&1 &  //默认端口为9000，可自定义指定
ss -tnulp|grep 9107
//tcp LISTEN 0 50 :::9107 :::* users:(("java",pid=28816,fd=116))

#访问验证
10.0.0.80:9107

kafka-mamager测试

如果没有在 Kafka 中配置过 JMX_PORT，不要选择第一个复选框。Enable JMX Polling如果选择了该复选框，Kafka-manager 可能会无法启动

认识kafka

kafka了解

kafka是一款性能非常好的并且支持分布式的消息队列中间件。由于它的高吞吐特性，Kafka通常使用在大数据领域，如日志收集平台。其实Kafka是一个流处理平台，这个概念不太好理解，之所以叫做流，是因为它在工作中就像是一个可以支撑高吞吐量的管道，数据像水一样流进去，然后另外一端再去读取这些数据。我们就可以把Kafka看作是一种特殊的消息队列中间件。
kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。 kafka集群几乎不需要维护任何consumer和producer状态信息,这些信息有zookeeper保存;因此producer和consumer的客户端实现非常轻量级,它们可以随意离开,而不会对集群造成额外的影响。
Kafka与传统消息系统相比，有以下不同：

1)它被设计为一个分布式系统，易于向外扩展；
2)它同时为发布和订阅提供高吞吐量；
3)它支持多订阅者，当失败时能自动平衡消费者；
4)它将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。

Kafka中有几个关键角色和概念

1）Producer

消息生产者，是消息的产生源头，负责生成消息并发送给Kafka。
生产者创建消息。在其他发布与订阅系统中，生产者可能被称为发布者或写入者。一般情况下，一个消息会被发布到一个特定的主题上。生产者在默认情况下把消息均衡地分布到主题的所有分区上，而并不关心特定消息会被写到哪个分区。不过，在某些情况下，生产者会把消息直接写到指定的分区。这通常是通过消息键和分区器来实现的，分区器为键生成一个散列值，并将其映射到指定的分区上。这样可以保证包含同一个键的消息会被写到同一个分区上。生产者也可以使用自定义的分区器，根据不同的业务规则将消息映射到分区。

2）Consumer

消息消费者，是消息的使用方，负责消费Kafka服务器上的消息。
消费者读取消息。在其他发布与订阅系统中，消费者可能被称为订阅者或读者。消费者订阅一个或多个主题，并按照消息生成的顺序读取它们。消费者通过检查消息的偏移量来区分已经读取过的消息。偏移量是另一种元数据，它是一个不断递增的整数值，在创建消息时，Kafka 会把它添加到消息里。在给定的分区里，每个消息的偏移量都是唯一的。消费者把每个分区最后读取的消息偏移量保存在 Zookeeper 或 Kafka 上，如果消费者关闭或重启，它的读取状态不会丢失。
消费者是消费者群组的一部分，也就是说，会有一个或多个消费者共同读取一个主题。群组保证每个分区只能被一个消费者使用。

3）Topic

主题，由用户自定义，并配置在Kafka服务器，用于建立生产者和消费者之间的订阅关系，生产者将消息发送到指定的Topic，然后消费者再从该Topic下去取消息。
 一个Topic可以认为是一类消息，每个topic将被分成多个partition(区),每个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型数字，它是唯一标记一条消息。它唯一的标记一条消息。kafka并没有提供其他额外的索引机制来存储offset，因为在kafka中几乎不允许对消息进行“随机读写”。

4）Partition

消息分区，一个Topic下面会有多个Partition，每个Partition都是一个有序队列，Partition中的每条消息都会被分配一个有序的id。
物理上的分区，topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。
Kafka 通过分区来实现数据冗余和伸缩性。分区可以分布在不同的服务器上。

5）Broker

    这个其实就是Kafka服务器了，无论是单台Kafka还是集群，被统一叫做Broker。
    一个独立的 Kafka 服务器被称为 broker。broker 接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。broker 为消费者提供服务，对读取分区的请求作出响应，返回已经提交到磁盘上的消息。根据特定的硬件及其性能特征，单个 broker 可以轻松处理数千个分区以及每秒百万级的消息量。
Kafka 集群包含一个或多个服务器，服务器节点称为broker。
broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。
如果某topic有N个partition，集群有(N M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。
如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。
    broker 是集群的组成部分。每个集群都有一个 broker 同时充当了集群控制器的角色（自动从集群的活跃成员中选举出来）。控制器负责管理工作，包括将分区分配给 broker 和监控broker。在集群中，一个分区从属于一个 broker，该 broker 被称为分区的首领。一个分区可以分配给多个 broker，这个时候会发生分区复制（见图 ）。这种复制机制为分区提供了消息冗余，如果有一个 broker 失效，其他 broker 可以接管领导权。不过，相关的消费者和生产者都要重新连接到新的首领。

6）Group

消费者分组，将同一类的消费者归类到一个组里。在Kafka中，多个消费者共同消费一个Topic下的消息，每个消费者消费其中的部分消息，这些消费者就组成了一个分组，拥有同一个组名。
kafka通过zookeeper管理集群配置，选举leader

kafka特点

1）Kafka：内存、磁盘、数据库、支持大量堆积

kafka的最小存储单元是分区，一个topic包含多个分区，kafka创建主题时，这些分区会被分配在多个服务器上，通常一个broker一台服务器。

分区首领会均匀地分布在不同的服务器上，分区副本也会均匀的分布在不同的服务器上，确保负载均衡和高可用性，当新的broker加入集群的时候，部分副本会被移动到新的broker上。

根据配置文件中的目录清单，kafka会把新的分区分配给目录清单里分区数最少的目录。

默认情况下，分区器使用轮询算法把消息均衡地分布在同一个主题的不同分区中，对于发送时指定了key的情况，会根据key的hashcode取模后的值存到对应的分区中。

2）Kafka：支持负载均衡
a)一个broker通常就是一台服务器节点。对于同一个Topic的不同分区，Kafka会尽力将这些分区分布到不同的Broker服务器上，zookeeper保存了broker、主题和分区的元数据信息。分区首领会处理来自客户端的生产请求，kafka分区首领会被分配到不同的broker服务器上，让不同的broker服务器共同分担任务。
每一个broker都缓存了元数据信息，客户端可以从任意一个broker获取元数据信息并缓存起来，根据元数据信息知道要往哪里发送请求。
b)kafka的消费者组订阅同一个topic，会尽可能地使得每一个消费者分配到相同数量的分区，分摊负载。
c)当消费者加入或者退出消费者组的时候，还会触发再均衡，为每一个消费者重新分配分区，分摊负载。
kafka的负载均衡大部分是自动完成的，分区的创建也是kafka完成的，隐藏了很多细节，避免了繁琐的配置和人为疏忽造成的负载问题。
d)发送端由topic和key来决定消息发往哪个分区，如果key为null，那么会使用轮询算法将消息均衡地发送到同一个topic的不同分区中。如果key不为null，那么会根据key的hashcode取模计算出要发往的分区。
3）集群方式，天然的‘Leader-Slave’无状态集群，每台服务器既是Master也是Slave。
分区首领均匀地分布在不同的kafka服务器上，分区副本也均匀地分布在不同的kafka服务器上，所以每一台kafka服务器既含有分区首领，同时又含有分区副本，每一台kafka服务器是某一台kafka服务器的Slave，同时也是某一台kafka服务器的leader。
kafka的集群依赖于zookeeper，zookeeper支持热扩展，所有的broker、消费者、分区都可以动态加入移除，而无需关闭服务，与不依靠zookeeper集群的mq相比，这是最大的优势。

Kafka工作流程

1）生产者定期向主题发送消息。

2）Kafka broker将所有消息存储在为该特定主题配置的分区中。它确保消息在分区之间平等共享。如果生产者发送两个消息，并且有两个分区，则Kafka将在第一个分区中存储一个消息，在第二个分区中存储第二个消息。

3）消费者订阅一个特定的主题。

4）一旦消费者订阅了一个主题，Kafka将向消费者提供该主题的当前偏移量，并将偏移量保存在ZooKeeper中。

5）消费者将定期请求Kafka新消息。

6）一旦Kafka收到来自生产者的消息，它会将这些消息转发给消费者。

7）消费者将收到消息并处理它。

8）一旦消息被处理，消费者将向Kafka broker发送确认。

9）一旦Kafka收到确认，它会将偏移量更改为新值，并在ZooKeeper中进行更新。由于ZooKeeper中保留了偏移量，因此即使在服务器出现故障时，消费者也可以正确读取下一条消息。

这篇好文章是转载于：学新通技术网