大家好，欢迎来到IT知识分享网。

作者：ninetyhe，腾讯后台开发工程师

| 导语 温故而知新，反复学习优秀的框架，定有所获。因为工作原因，需要用到Kafka的特殊场景，周末再次阅读了kafka的资料，收获不少。

kafka由LinkedIn公司推出的一个高吞吐的分布式消息系统，通俗的说就是一个基于发布和订阅的消息队列，官网地址：https://kafka.apache.org/intro

应用场景

• 异步解构：在上下游没有强依赖的业务关系或针对单次请求不需要立刻处理的业务
• 系统缓冲：有利于解决服务系统的吞吐量不一致的情况，尤其对处理速度较慢的服务来说起到缓冲作用
• 消峰作用：对于短时间偶现的极端流量，对后端的服务可以启动保护作用
• 数据流处理：集成spark做实事数据流处理

Kafka拓扑图（多副本机制）

由上图我们可以发现Kafka是分布式，同时对于每一个分区都存在多副本，同时整个集群的管理都通过zookeeper管理。

Kafka核心组件

broker

Kafka 服务器，负责消息存储和转发；一broker就代表一个kafka节点。一个broker可以包含多个topic

topic

消息类别，Kafka 按照 topic 来分类消息

partition

• topic 的分区，一个 topic 可以包含多个 partition，topic 消息保存在各个partition 上；由于一个topic能被分到多个分区上，给kafka提供给了并行的处理能力，这也正是kafka高吞吐的原因之一。
• partition 物理上由多个 segment 文件组成，每个 segment 大小相等，顺序读写（这也是kafka比较快的原因之一，不需要随机写）。每个Segment 数据文件以该段中最小的 offset ，文件扩展名为.log。当查找 offset 的 Message的时候，通过二分查找快找到Message所处于的Segment中。

offset

• 消息在日志中的位置，可以理解是消息在 partition 上的偏移量，也是代表该消息的唯一序号。
• 同时也是主从之间的需要同步的信息

Producer

生产者，负责向Kafka Broker发消息的客户端

Consumer

消息消者，负责消费Kafka Broker中的消息

Consumer Group

消费者组，每个 Consumer 必须属于一个 group；（注意的是 一个分区只能由组内一个消费者消费，消费者组之间互不影响。）

Zookeeper

管理kafka集群，负责存储了集群 broker、topic、partition 等 meta数据存储，同时也负责broker故障发现，partition leader 选举，负载均衡等功能。

服务治理

既然Kafka是分布式的发布/订阅系统，这样如果做的集群之间数据同步和一致性，kafka是不是肯定不会丢消息呢？以及宕机的时候如果进行Leader选举呢？

数据同步

在Kafka中的Partition有一个leader与多个follower，producer往某个Partition中写入数据是，只会往leader中写入数据，然后数据才会被复制进其他的Replica中。而每一个follower可以理解成一个消费者，定期去leader去拉去消息。而只有数据同步了后，kafka才会给生产者返回一个ACK告知消息已经存储落地了。

ISR

在Kafka中，为了保证性能，Kafka不会采用强一致性的方式来同步主从的数据。而是维护了一个：in-sync Replica的列表，Leader不需要等待所有Follower都完成同步，只要在ISR中的Follower完成数据同步就可以发送ack 给生产者即可认为消息同步完成。同时如果发现ISR里面某一个follower落后太多的话，就会把它剔除。

具体流程如下：

上述的做法并无法保证kafka一定不丢消息。虽然Kafka 通过多副本机制中最大限度保证消息不会丢失，但是如果数据已经写入系统 page cache 中但是还没来得及刷入磁盘，此时突然机器宕机或者掉电，那消息自然而然的就会丢失。

Kafka故障恢复

Kafka通过Zookeeper连坐集群的管理，所以这里的选举机制采用的是Zab(zookeeper使用)。

• 生产者发生消息给leader，这个时候leader完成数据存储，突然发生故障，没有给producer返回ack；
• 通过ZK选举，其中一个follower成为leader，这个时候producer重新请求新的leader，并存储数据

Kafka为什么这么快

顺序写磁盘

Kafka采用了顺序写磁盘，而由于顺序写磁盘相对随机写，减少了寻地址的耗费时间。（在Kafka的每一个分区里面消息是有序的

Page Cache

Kafka在OS系统方面使用了Page Cache而不是我们平常所用的Buffer。Page Cache其实不陌生，也不是什么新鲜事物。

我们在linux上查看内存的时候，经常可以看到buff/cache，两者都是用来加速IO读写用的，而cache是作用于读，也就是说，磁盘的内容可以读到cache里面这样，应用程序读磁盘就非常快；而buff是作用于写，我们开发写磁盘都是，一般如果写入一个buff里面再flush就非常快。而kafka正是把这两者发挥了极致：Kafka虽然是scala写的，但是依旧在Java的虚拟机上运行，尽管如此，kafka它还是尽量避开了JVM的限制，它利用了Page cache来存储，这样躲开了数据在JVM因为GC而发生的STW。另一方面也是Page Cache使得它实现了零拷贝，具体下面会讲。

零拷贝

无论是优秀的Netty还是其他优秀的Java框架，基本都在零拷贝减少了CPU的上下文切换和磁盘的IO。当然Kafka也不例外。零拷贝的概念具体这里不作太详细的复述，大致的给大家讲一下这个概念。

传统的一次应用程请求数据的过程

这里大致可以发传统的方式发生了4次拷贝，2次DMA和2次CPU，而CPU发生了4次的切换。（DMA 简单理解就是，在进行 I/O 设备和内存的数据传输的时候，数据搬运的工作全部交给 DMA 控制器，而 CPU 不再参与任何与数据搬运相关的事情）

零拷贝的方式

通过优化我们可以发现，CPU只发生了2次的上下文切换和3次数据拷贝。（linux系统提供了系统事故调用函数“ sendfile()”，这样系统调用，可以直接把内核缓冲区里的数据拷贝到 socket 缓冲区里，不再拷贝到用户态）

分区分段

我们上面也介绍过了，kafka采取了分区的模式，而每一个分区又对应到一个物理分段，而查找的时候可以根据二分查找快速定位。这样不仅提供了数据读的查询效率，也提供了并行操作的方式

数据压缩

Kafka对数据提供了：Gzip和Snappy压缩协议等压缩协议，对消息结构体进行了压缩，一方面减少了带宽，也减少了数据传输的消耗

Kafka安装

安装JDK

由于使用压缩包还需要自己配置环境变量，所以这里推荐直接用yum安装，熟悉查看目前Java的版本：

yum -y list Java*

IT知识分享网

安装你想要的版本，这里我是1.8

IT知识分享网yum install java-1.8.0-openjdk-devel.x86_64

查看是否安装成功

Java -version

安装Zookeeper

首先需要去官网下载安装包，然后解压

IT知识分享网tar -zxvf zookeeper-3.4.9.tar.gz  

要做的就是将这个文件复制一份，并命名为：zoo.cfg，然后在zoo.cfg中修改自己的配置即可

cp zoo_sample.cfg zoo.cfg
vim zoo.cfg

主要配置解释如下

# zookeeper内部的基本单位，单位是毫秒，这个表示一个tickTime为2000毫秒，在zookeeper的其他配置中，都是基于tickTime来做换算的
tickTime=2000
# 集群中的follower服务器(F)与leader服务器(L)之间 初始连接 时能容忍的最多心跳数（tickTime的数量）。
initLimit=10
#syncLimit：集群中的follower服务器(F)与leader服务器(L)之间 请求和应答 之间能容忍的最多心跳数（tickTime的数量）
syncLimit=5
# 数据存放文件夹，zookeeper运行过程中有两个数据需要存储，一个是快照数据（持久化数据）另一个是事务日志
dataDir=/tmp/zookeeper
## 客户端访问端口
clientPort=2181

配置环境变量

vim ~/.bash_profile
export ZK=/usr/local/src/apache-zookeeper-3.7.0-bin
export PATH=$PATH:$ZK/bin
export PATH 
// 启动
zkServer.sh start

下面能看启动成功

安装Kafka

下载kafka

https://www.apache.org/dyn/closer.cgi?path=/kafka/2.8.0/kafka-2.8.0-src.tgz

安装kafka

 tar -xzvf kafka_2.12-2.0.0.tgz 

配置环境变量

 export ZK=/usr/local/src/apache-zookeeper-3.7.0-bin
 export PATH=$PATH:$ZK/bin
 export KAFKA=/usr/local/src/kafka
 export PATH=$PATH:$KAFKA/bin

启动Kafka

 nohup kafka-server-start.sh 自己的配置文件路径/server.properties &

大功告成！

参考资料

《深入理解Kafka：核心设计实践原理》