Kafka原理及Kafka群集部署

博文大纲：
一、Kafka概述
1）消息队列
2）为什么要使用消息队列？
3）什么是Kafka？
4）Kafka的特性
5）Kafka架构
6）Topic和Partition的区别
7）kafka流程图
8）Kafka的文件存储机制
9）数据的可靠性和持久性保证
10）leader选举
二、部署单机Kafka
1）部署Kafka
2）测试Kafka
三、部署Kafka群集
1）环境准备
2）部署zookeeper群集
3）部署Kafka群集

一、Kafka概述

1）消息队列

1）点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）

点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此；

2）发布/订阅模式（一对多，数据生产后，推送给所有订阅者）

发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。

2）为什么要使用消息队列？

1）解耦

允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

2）冗余

消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

3）扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。

4）灵活性&&削峰填谷

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5）可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

6）顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka 保证一个 Partition 内的消息的有序性）

7）缓冲

有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

8）异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

3）什么是Kafka？

kafka是由Apache软件基金会发布的一个开源流处理平台，由Scala和Java编写。它是一种高吞吐量的分布式发布的订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

4）Kafka的特性

kafka是一种高吞吐量的分布式发布订阅消息系统，具有以下特性：
1）通过磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能；
2）持久性：使用文件性存储，日志文件存储消息，需要写入硬盘，采用达到一定阈值才写入硬盘，从而减少磁盘I/O，如果kafka突然宕机，数据会丢失一部分；
3）高吞吐量：即使是非常普通的硬件kafka也可以支持每秒数百万的消息；
4）支持通过kafka服务器和消费机集群来分区消息；
5）支持Hadoop并行数据加载；

5）Kafka架构

架构图中的各个组件作用：
1）Producer ：消息生产者，就是向 kafka broker 发消息的客户端；
2）Broker ：一台 kafka 服务器就是一个 broker。一个集群由多个broker 组成。一个 broker可以容纳多个 topic；
3）Consumer ：消息消费者，向 kafka broker 取消息的客户端；
4）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。partition 中的每条消息都会被分配一个有序的 id（offset）。kafka 只保证按一个 partition 中的顺序将消息发给consumer，不保证一个 topic 的整体（多个 partition 间）的顺序；
5）Topic ：Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic；
6）ConsumerGroup：每个Consumer属于一个特定的Consumer Group，一条消息可以发送到多个不同的Consumer Group，但是一个Consumer Group中只能有一个Consumer能够消费该消息；

6）Topic和Partition的区别

一个topic可以认为一个一类消息，每个topic将被分成多个partition，每个partition在存储层面是append log文件。任何发布到此partition的消息都会被追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为long型的数字，它唯一标记一条消息。每条消息都被append到partition中，是顺序写磁盘，因此效率非常高（顺序写磁盘比随机写内存的速度还要高，这是kafka高吞吐率的一个很重要的保证）。

每一条消息被发送到broker中，会根据partition规则选择被存储到哪一个partition（默认采用轮询的方式进行写入数据）。如果partition规则设置合理，所有消息可以均匀分布到不同的partition里，这样就实现了水平扩展。（如果一个topic对应一个文件，那这个文件所在的机器I/O将会成为这个topic的性能瓶颈，而partition解决了这个问题），如果消息被消费则保留append.log两天。

7）kafka流程图

如上图所示，一个典型的kafka体系架构包括若干Producer（可以是服务器日志，业务数据，页面前端产生的page view等），若干个broker（kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer（Group），以及一个Zookeeper集群。kafka通过Zookeeper管理集群配置，选举出leader，以及在consumer group发生变化时进行重新调整。Producer使用push（推）模式将消息发布到broker，consumer使用pull（拉）模式从broker订阅并消费消息。

zookeeper群集中有两个角色：leader和follower，leader对外提供服务，follower负责leader里面所产生内容同步消息写入生成时产生replicas（副本）；

kafka的高可靠性的保证来源于其健壮的副本（replicas）策略。通过调节其副本相关参数，可以使得kafka在性能和可靠性之间运转之间的游刃有余。kafka从0.8.x版本开始提供partition级别的复制的。

8）Kafka的文件存储机制

kafka中消息是以topic进行分类的，生产者通过topic向kafka broker发送消息，消费者通过topic读取数据。然而topic在物理层面又能以partition为分组，一个topic可以分为若干个partition，partition还可以细分为segment，一个partition物理上由多个segment组成。

为了便于说明问题，假设这里只有一个kafka集群，且这个集群只有一个kafka broker，也就是只有一台物理机。在这个kafka broker的server.properties配置文件中定义kafka的日志文件存放路径以此来设置kafka消息文件存储目录，与此同时创建一个topic：test，partition的数量为4，启动kafka就可以在日志存放路径中看到生成4个目录，在kafka文件存储中，同一个topic下有多个不同的partition，每个partition为一个目录，partition的名称规则为：topic名称+有序序号，第一个序号从0开始。

segment是什么？

如果就以partition为最小存储单位，我们可以想象当Kafka producer不断发送消息，必然会引起partition文件的无限扩张，这样对于消息文件的维护以及已经被消费的消息的清理带来严重的影响，所以这里以segment为单位又将partition细分。每个partition(目录)相当于一个巨型文件被平均分配到多个大小相等的segment(段)数据文件中（每个segment 文件中消息数量不一定相等）这种特性也方便old segment的删除，即方便已被消费的消息的清理，提高磁盘的利用率。每个partition只需要支持顺序读写就行。

segment文件由两部分组成，分别为".index"文件和".log"文件，分别表示为segment索引文件和数据文件。这两个文件的命令规则为：partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值（偏移量），数值大小为64位，20位数字字符长度，没有数字用0填充。

9）数据的可靠性和持久性保证

当producer向leader发送数据时，可以通request.required.acks参数来设置数据可靠性的级别：

• 1（默认）：producer的leader已成功收到数据并得到确认。如果leader宕机了，则会丢失数据；
• 0 ：producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的；
• -1：producer需要等待所有follower都确认接收到数据后才算一次发送完成，可靠性最高；

10）leader选举

一条消息只有被所有follower都从leader复制过去才会被认为已提交。这样就避免了部分数据被写进了leader，还没来得及被任何follower复制就宕机了，而造成数据丢失。而对于producer而言，它可以选择是否等待消息commit。

一种非常常用的选举leader的方式是"少数服从多数"，在进行数据的复制过程中，存在多个follower，并且每个follower的数据速度都不相同，当leader宕机后，当前的follower上谁的数据最多谁就是leader。

二、部署单机Kafka

1）部署Kafka

Kafka服务默认需要依赖Java环境，本次使用Centos 7系统，默认已经具备Java的环境。

Kafka群集所需软件下载链接：https://pan.baidu.com/s/1VHUH-WsptDB2wOugxvfKVg
提取码：47x7

[root@kafka ~]# tar zxf kafka_2.11-2.2.1.tgz [root@kafka ~]# mv kafka_2.11-2.2.1 /usr/local/kafka[root@kafka ~]# cd /usr/local/kafka/bin/[root@kafka bin]# ./zookeeper-server-start.sh ../config/zookeeper.properties &#启动zookeeper，启动时需指定zookeeper的配置文件#需添加"&"符号，表示后台运行，否则会占用前台的终端[root@kafka bin]#  ./kafka-server-start.sh ../config/server.properties &#启动kafka，启动方式与启动zookeeper一样[root@kafka bin]# netstat -anpt | grep 9092#kafka的监听端口为9092，确认端口在监听的状态

由于kafka是通过zookeeper来调度的，所以，即使是单机kafka也需要启动zookeeper服务，kafka的安装目录下是默认集成了zookeeper的，直接启动即可。

2）测试Kafka

[root@kafka bin]# ./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test#在本机创建一个topic，名称为test，副本数量为1，分区数量为1[root@kafka bin]# ./kafka-topics.sh --list --bootstrap-server localhost:9092#查看本机的topic[root@kafka bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic test#向创建的topic中添加消息，消息内容自定义>aaaa>bbbb>cccc[root@kafka bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginningaaaabbbbcccc#开启新的终端，进行读取消息测试，"--from-beginning"表示从开头读取

测试成功！

三、部署Kafka群集

1）环境准备

2）部署zookeeper群集

1）主机kafka01配置

[root@kafka01 ~]# tar zxf zookeeper-3.4.9.tar.gz [root@kafka01 ~]# mv zookeeper-3.4.9 /usr/local/zookeeper#安装zookeeper[root@kafka01 ~]# cd /usr/local/zookeeper/conf[root@kafka01 conf]# cp -p zoo_sample.cfg zoo.cfg [root@kafka01 conf]#  sed -i 's/dataDir=\/tmp\/zookeeper/dataDir=\/usr\/local\/zookeeper\/data/g' zoo.cfg[root@kafka01 conf]#  echo "server.1 192.168.1.6:2888:3888" >> zoo.cfg [root@kafka01 conf]#  echo "server.2 192.168.1.7:2888:3888" >> zoo.cfg [root@kafka01 conf]#  echo "server.3 192.168.1.8:2888:3888" >> zoo.cfg  [root@kafka01 conf]# egrep -v '^$|^#' zoo.cfg        #更改后的配置文件tickTime=2000            #节点之间的心跳检测时间单位为毫秒 initLimit=10      #节点之间检查失败次数超过后断开相应的节点syncLimit=5      #达到5个访问进行同步数据dataDir=/usr/local/zookeeper/data    #日志文件存放路径clientPort=2181              #监听端口server.1 192.168.1.6:2888:3888server.2 192.168.1.7:2888:3888server.3 192.168.1.8:2888:3888#声明参与集群的主机，2888和3888端口用于群集内部通信[root@kafka01 conf]# mkdir /usr/local/zookeeper/data[root@kafka01 conf]# echo 1 > /usr/local/zookeeper/data/myid#创建所需目录及设置节点的ID号[root@kafka01 conf]# scp -r /usr/local/zookeeper/ root@192.168.1.7:/usr/local/[root@kafka01 conf]# scp -r /usr/local/zookeeper/ root@192.168.1.8:/usr/local/#将配置好的zookeeper目录复制到群集内的其他节点[root@kafka01 conf]# /usr/local/zookeeper/bin/zkServer.sh start[root@kafka01 conf]# netstat -anpt | grep 2181#确认zookeeper服务端口正在监听

2）主机kafka02配置

[root@kafka02 ~]# echo 2 > /usr/local/zookeeper/data/myid #修改ID号为2[root@kafka02 ~]# /usr/local/zookeeper/bin/zkServer.sh start#启动zookeeper

3）主机kafka03配置

[root@kafka03 ~]#  echo 3 > /usr/local/zookeeper/data/myid#修改ID号为3[root@kafka03 ~]#  /usr/local/zookeeper/bin/zkServer.sh start#启动zookeeper

4）查看zookeeper群集内节点的角色

[root@kafka01 conf]#  /usr/local/zookeeper/bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /usr/local/zookeeper/bin/../conf/zoo.cfgMode: follower         #角色为follower[root@kafka02 ~]#  /usr/local/zookeeper/bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /usr/local/zookeeper/bin/../conf/zoo.cfgMode: follower         #角色为follower[root@kafka03 ~]#  /usr/local/zookeeper/bin/zkServer.sh statusZooKeeper JMX enabled by defaultUsing config: /usr/local/zookeeper/bin/../conf/zoo.cfgMode: leader      #角色为leader

3）部署Kafka群集

1）主机kafka01配置

[root@kafka01 ~]# tar zxf kafka_2.11-2.2.1.tgz [root@kafka01 ~]# mv kafka_2.11-2.2.1 /usr/local/kafka#安装kafka[root@kafka01 ~]# cd /usr/local/kafka/config/[root@kafka01 config]# sed -i 's/broker.id=0/broker.id=1/g' server.properties [root@kafka01 config]# sed -i 's/#listeners=PLAINTEXT:\/\/:9092/listeners=PLAINTEXT:\/\/192.168.1.6:9092/g' server.properties[root@kafka01 config]# sed -i 's/#advertised.listeners=PLAINTEXT:\/\/your.host.name:9092/advertised.listeners=PLAINTEXT:\/\/192.168.1.6:9092/g' server.properties[root@kafka01 config]# sed -i 's/log.dirs=\/tmp\/kafka-logs/log.dirs=\/usr\/local\/zookeeper\/data/g' server.properties[root@kafka01 config]#  sed -i 's/zookeeper.connect=localhost:2181/zookeeper.connect=192.168.1.6:2181,192.168.1.7:2181,192.168.1.8:2181/g' server.properties[root@kafka01 config]#  sed -i 's/zookeeper.connection.timeout.ms=6000/zookeeper.connection.timeout.ms=600000/g' server.properties#修改kafka的配置文件[root@kafka01 config]# egrep -v '^$|^#' server.properties#修改后的配置文件broker.id=1     #kafka的ID号，这里为1，其他节点依次是2、3listeners=PLAINTEXT://192.168.1.6:9092   #节点监听地址，填写每个节点自己的IP地址advertised.listeners=PLAINTEXT://192.168.1.6:9092    #集群中节点内部交流使用的端口，填写每个节点自己的IP地址num.network.threads=3num.io.threads=8socket.send.buffer.bytes=102400socket.receive.buffer.bytes=102400socket.request.max.bytes=104857600log.dirs=/usr/local/zookeeper/datanum.partitions=1num.recovery.threads.per.data.dir=1offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1log.retention.hours=168log.segment.bytes=1073741824log.retention.check.interval.ms=300000zookeeper.connect=192.168.1.62:2181,192.168.1.7:2181,192.168.1.8:2181#声明链接zookeeper节点的地址zookeeper.connection.timeout.ms=600000#修改这的时间，单位是毫秒，为了防止连接zookeeper超时group.initial.rebalance.delay.ms=0[root@kafka01 config]# scp -r /usr/local/kafka/ root@192.168.1.7:/usr/local/[root@kafka01 config]# scp -r /usr/local/kafka/ root@192.168.1.8:/usr/local/#将修改后的kafka目录发送至其他节点[root@kafka01 bin]# ./kafka-server-start.sh ../config/server.properties &#启动kafka[root@kafka01 bin]# netstat -anpt | grep 9092确认端口在监听

2）主机kafka02配置

[root@kafka02 ~]# cd /usr/local/kafka/[root@kafka02 kafka]#  sed -i 's/broker.id=1/broker.id=2/g' config/server.properties[root@kafka02 kafka]# sed -i 's/192.168.1.6:9092/192.168.1.7:9092/g' config/server.properties #修改与kafka01冲突之处[root@kafka02 kafka]# cd bin/[root@kafka02 bin]#  ./kafka-server-start.sh ../config/server.properties &#启动kafka[root@kafka02 bin]# netstat -anpt | grep 9092#确认端口在监听

3）主机kafka03配置

[root@kafka03 ~]# cd /usr/local/kafka/[root@kafka03 kafka]# sed -i 's/broker.id=1/broker.id=3/g' config/server.properties[root@kafka03 kafka]# sed -i 's/192.168.1.6:9092/192.168.1.8:9092/g' config/server.properties#修改与kafka01冲突之处[root@kafka03 kafka]# cd bin/[root@kafka03 bin]#  ./kafka-server-start.sh ../config/server.properties &#启动kafka[root@kafka03 bin]# netstat -anpt | grep 9092#确认端口在监听

4）发布和订阅消息测试

[root@kafka01 bin]#  ./kafka-topics.sh --create --bootstrap-server 192.168.1.6:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic#创建名为my-replicated-topic的topic[root@kafka01 bin]# ./kafka-topics.sh --describe --bootstrap-server 192.168.1.6:9092 --topic my-replicated-topic#查看topic的状态和leaderTopic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:segment.bytes=1073741824#返回的信息表示partition数量为1，副本数量为3，segment字节数为1073741824    Topic: my-replicated-topic  Partition: 0    Leader: 1   Replicas: 1,3,Isr: 1,3,2#名称为"my-replicated-topic"，ID为1的节点为leader[root@kafka01 bin]#  ./kafka-console-producer.sh --broker-list 192.168.1.6:9092 --topic my-replicated-topic#向名为my-replicated-topic的topic 中插入数据进行测试>aaaaaaaa>bbbbbbbbbbbbbbb>ccccccccccccccccccccccc[root@kafka02 bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.1.7:9092 --from-beginning --topic my-replicated-topic#在其他节点上查看插入的数据aaaaaaaabbbbbbbbbbbbbbbccccccccccccccccccccccc

5）模拟leader宕机，查看topic的状态及新的leader

[root@kafka01 bin]# ./kafka-topics.sh --describe --bootstrap-server 192.168.1.6:9092 --topic my-replicated-topic#查看名为my-replicated-topic的topic状态Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:segment.bytes=1073741824    Topic: my-replicated-topic  Partition: 0    Leader: 1   Replicas: 1,3,Isr: 1,3,2#可以看出ID号为1的节点是leader[root@kafka01 bin]# ./kafka-server-stop.sh#ID号为1的节点停止kafka服务[root@kafka02 bin]#  ./kafka-topics.sh --describe --bootstrap-server 192.168.1.7:9092 --topic my-replicated-topic#在第二个节点上查看topic状态Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:segment.bytes=1073741824    Topic: my-replicated-topic  Partition: 0    Leader: 3   Replicas: 1,3,Isr: 3,2#可以看出leader换成了ID为3的节点

-------- 本文至此结束，感谢阅读 --------