kafka架构
一个典型的kafka集群包含若干个Producer(可以是应用节点产生的消息，也可以是通过Flume收集日志产生的事件)、若干个Broker(kafka支持水平扩展)、若干个Consumer Group以及zookeeper集群。kafka通过zookeeper管理集群配置及服务协同。Producer使用push模式将消息发布到broker，consumer通过监听使用pull模式从broker订阅并消费消息。
多个broker协同工作，producer和consumer部署在各个业务逻辑中。三者通过zookeeper管理协调请求和转发。这样就组成了一个高性能的分布式消息发布和订阅系统。
下图又一个细节是和其他MQ中间件不同的点，Producer发送消息到broker的过程是push，而consumer从borker消费消息的过程是pull，主动去拉数据。而不是broker把数据主动发给consumer。
名词解释
1、borker
kafka集群包含一个或多个服务器，这种服务器被称为broker。broker端不维护数据的消费状态，提升了性能。直接使用磁盘进行存储，线性读写，速度快；避免了数据在jvm内存和系统内存的复制，减少耗性能的创建对象和垃圾回收。
2、Producer
复制发布消息到kafka broker
3、consumer
消息消费者，向kakfa broker读取消息的客户端，consumer从broker拉取(pull)数据并进行处理。
4、Topic
每条发布到kafka集群的消息都有一个类别，这个类别被称为Topic。(物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需要指定消息的Topic即可生产或消费数据而不必关系数据存于何处)
5、Partition
Partition是物理上的概念，每个Topic包含一个或多个Partition。
6、consumer group
每个Consumer属于一个特定的Consummer Group(可为每个Consumer指定group name，若不指定group name则属于默认的group)
7、Topic & Partition
Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使得kafka的吞吐率可以线性提高，物理上把一个Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partiton的所有消息和索引文件。若创建topic1和topic2两个topic，且分别有13个19个分区，则整个集群上会相应生成共32个文件夹(本文所用集群共8个节点，此处topic1和topic2 replication-factor均为1)
kafka安装部署
下载kafka
https://archive.apache.org/dist/kafka/2.0.0/kafka_2.11-2.0.0.tgz
安装过程
安装过程非常简单，只需要解压就行，因为这个是编译好之后的可执行程序
tar -zxvf kafka_2.11-2.0.0.tgz 解压
配置zookeeper
因为kafka依赖于zookeeper来做master选举一起其他数据的维护，所以需要先启动zookeeper节点
kafka内置了zookeeper的服务，所以在bin目录下提供了这些脚本
zookeeper-server-start.sh
zookeeper-server-stop.sh
在config目录下，存在一些配置文件
zookeeper.properties
server.properties
所以我们可以通过下面的脚本来启动zk服务，当然，也可以自己搭建zk的集群来实现
sh zookeeper-server-start.sh -daemon ../config/zookeeper.properties
启动和停止kafka
修改server.properties, 增加zookeeper的配置
zookeeper.connect=localhost:2181
启动kafka
sh kafka-server-start.sh -damoen config/server.properties
停止kafka
sh kafka-server-stop.sh -daemon config/server.properties
kafka的基本操作
创建topic
sh kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --
partitions 1 --topic test
Replication-factor 表示该topic需要在不同的broker中保存几份，这里设置成1，表示在两个broker中
保存两份
Partitions 分区数
查看topic
sh kafka-topics.sh --list --zookeeper localhost:2181
查看topic属性
sh kafka-topics.sh --describe --zookeeper localhost:2181 --topic first_topic
消费消息
sh kafka-console-consumer.sh --bootstrap-server 192.168.13.106:9092 --topic test
--from-beginning
发送消息
sh kafka-console-producer.sh --broker-list 192.168.244.128:9092 --topic
first_topic
集群环境安装
环境准备
准备三台虚拟机
分别把kafka的安装包部署在三台机器上
以下配置修改均为server.properties
1、分别修改三台机器的server.properties配置，同一个集群中的每个机器的id必须唯一
broker.id=0
broker.id=1
broker.id=22、修改zookeeper的连接配置
zookeeper.connect=192.168.13.106:2181
3、修改listeners配置
如果配置了listeners，那么消息生产者和消费者会使用listeners的配置来进行消息的收发，否则，
会使用localhostPLAINTEXT表示协议，默认是明文，可以选择其他加密协议
listeners=PLAINTEXT://192.168.13.102:9092
4、分别启动三台服务器
sh kafka-server-start.sh -daemon ../config/server.properties
如何处理所有的Replica不工作的情况
在ISR中至少有一个follower时，kafka可以确保已经commit的数据不丢失，但如果某个Partition的所有Replica都宕机了，就无法保证数据不丢失
1、等待ISR中的任一个Replica"活过来"，并且选它作为Leader
2、选择第一个"活"过来的Replica(不一定是ISR中的)作为Leader
这就需要在可用性和一致性中做出一个简单的折衷。
如果一定要等待ISR中的Replica"活"过来，那不可用的时间就可能会相对较长。而且如果ISR中所有Replice都无法活过来，或者数据都丢失了，这个partition将永远不可用。
选择第一个活过来的Replica作为Leader，而这个Replica不是ISA中的Replica，那即使它并不保证已经包含了所有已commit的消息，它也会成为Leader而作为Consumer的数据源(前文有说明，所有读写都由Leader完成)。
副本数据同步原理
了解了副本的协同过程之后，还有一个最重要的机制，就是数据的同步过程。它需要解决：
1、怎么传播消息
2、在向消息发送端返回ack之前需要保证多少个Replica已经接收到这个消息
数据的处理过程是
下图中，深红色部分表示test_replica分区的leader副本，另外两个节点上浅色部分表示follower副本
Producer在发布消息到某个Partition时
1、先通过Zookeeper找到该Partition的Leader get/brokers/topics/<topic>/partitions/2/state，然后无论该Topic的Replication Factor为多少(也即该Paratiton有多少个Replica)，Producer只将该消息发送到该Partition的Leader。
2、Leader会将该消息写入其他本地Log，每个Follower都从Leader pull数据。这种方式上，Follower存储的数据顺序于Leader保持一致。
3、Follower在收到该消息并写入其Log后，向Leader发送ACK
4、一旦Leader收到了ISR中的所有Replica的ACK，该消息就被认为已经commit了，Leader将增加HW(HighWaterMark)并且向Producer发送ACK。
LEO：
即日志末端位移（log end offset），记录了该副本底层日志(log)中下一条消息的位移值。注意是下一条消息。也就是说，如果LEO=10，那么表示该副副本保存了10条消息，位移值范围是[0，9]。另外leader LEO和follower LEO的更新是有区别的。
HW(HighWatermark)：
即上面提到的水位置(Hight Water)。对于同一个副本对象而言，其HW值不会大于LEO值。小于等于HW值得多有消息都被认为是“已备份”的(replicated)。同理，leader副本和follower副本的HW更新是有区别的。
通过下面这幅图来表达LEO、HW的含义，随着follower副本不断和leader副本进行数据同步，follower副本你的LEO会逐渐后移并且追赶到leader副本，这个追赶上的判断标准是，当前副本的LEO是否大于或者等于leader副本的HW，这个追赶上也会使得被踢出的follower副本重新加入到ISR集合中。
另外，假如说下图的最右侧的follower副本被 提出ISR集合，也会导致这个分区HW发生变化，变成3
初始状态
初始状态下，leader和follower的HW和LEO都是0，leader副本会保存remote LEO，表示所有follower LEO，也会被被初始化为0，这个时候producer没有发送消息。follower还会不断地给leader发送FETCH请求，但是没有数据，这个请求会被leader寄存，当指定的时间之后会强制完成请求，这个时间配置是(replica.fetch.wait.max.ms)，如果在指定时间内producer有消息发过来，那么kafka会被唤醒fetch请求，让leader继续处理。
数据的同步处理会分两种情况，这两种情况下处理方式是不一样的
1、第一种是leader处理完producer请求之后，follower发送一个fetch请求过来
2、第二种是follower阻塞在leader指定时间之内，leader副本收到producer的请求。
第一种情况
生产者发送一条消息
leader处理完producer请求之后，follower发送一个fetch请求过来。状态图如下：
leader副本收到请求以后，会做几件事情
1、把消息追加到log文件，同事更新leader副本的LEO
2、尝试更新leader HW值。这个时候由于follower副本还没有发送fetch请求，那么leader的remote LEO仍然是0。leader会比较自己的LEO以及remote LEO的值发现最小值是0，与HW的值相同，所以不会更新HW。
follower fetch消息
follower发送fetch请求，leader副本的处理逻辑是
1、读取log数据、更新remote LEO=0(follower还没有写入这条消息，这个值是根据 follower的fetch请求中的offset来确定的)
2、尝试更新HW，因为这个时候LEO和remote LEO还是不一致，所以仍然是HW=0
3、把消息内容和当前分区的HW值发送给follower副本
follower副本收到response以后
1、将消息写入到本地log，同时更新follower的LEO
2、更新follower HW，本地的LEO和leader返回的HW进行比较取小的值，所以仍然是0
第一次交互结束以后，HW仍然是0，这个值会在下一次follower发起fetch请求时被更新
follower发第二次请求，leader收到请求以后
1、读取log数据
2、更新remote LEO=1，因为这次fetchh携带的offset是1
3、更新当前分区的HW，这个时候leader LEO和remote LEO都是1，所以HW的值也更新为1
4、把数据和当前分区的HW值返回给follower副本，这个时候如果没有数据，则返回空
follower副本收到response以后
1、如果又数据则写本地日志，并更新LEO
2、更新follower的HW值
到目前为止，数据的同步就完成了，意味着消费端能够消费offset=1这条消息。
第二种情况
前面所过，由于leader副本暂时没有数据过来，所以follower的fetch会被阻塞，直到等待超时或者leader接收到新的数据。当leader收到请求以后会被唤醒处于阻塞的fetch请求。处理过程基本上和前面说的一致
1、leader将消息写入本地日志，更新leader的LEO
2、唤醒follower的fetch请求
3、更新HW
kafka使用HW和LEO的方式来实现副本数据的同步，本身是一个好的设计，但是在这个地方会存在一个数据的丢失问题，当然这个丢失只出现在特定的背景下。我们回想一下，HW的值是在新的一轮FETCH中才会被更新。我们分析下这个过程为什么会出现数据丢失
数据丢失的问题
前提
min.insync.replicas=1 //设定ISR中的最小副本数是多少，默认值为1（在server.properties中配
置）， 并且acks参数设置为-1（表示需要所有副本确认）时，此参数才生效.
表达的含义是，至少需要多少个副本才能表示消息是提交的，所以，当min.insync.replicas=1的时候，一旦消息被写入leader端log即被认为是“已提交”，而延迟一轮FETCH RPC更新HW值设计使得follower HW值是异步延迟更新得，倘若在这个过程中leader发生变更，那么称为新的leader得follower的HW值就有可能是过期的 ，使得clients端认为是成功提交的消息被删除。
producer的ack
acks配置表示producer发送消息到broker上以后的确认值。有三个可选项
0：表示producer不要等待broker的消息确认。这个选项时延最小但同时风险最大(因为当server宕机时，数据将会丢失)
1：表示producer只需要获得kafka集群中的leader节点确认即可，这个选择时延较小同时确保了leader节点确认接收成功
all(-1)：需要ISR中所有的replica给予接收确认，速度最慢，安全性最高，但是由于ISR可能会缩小到仅包含一个Replica，所以设置参数为欸all不能一定避免数据丢失。
数据丢失的解决方案
在kafka0.11.0.0版本之后，引入了一个leader epoch来解决这个问题，所谓的leader epoch实际上是一对值(epoch，offset)，epoch代表leader版本号，从0开始递增，当leader发生过变更，epoch就+1，而offset则是对应这个epoch版本的leader写入第一条消息的offset，比如(0,0)，(1,50)，表示第一个leader从offset=0开始写消息，一共写了50条。第二个leader版本号是1，从offset=50开始写，这个信息会持久化在对应的分区的本地磁盘上，文件名是/tmp/kafka-log/topic/leader-epoch-checkpoint。
leader broker中会保存这样一个缓冲，并且定期写入到checkpoint文件中
当leader写log时它会尝试更新整个缓冲：如果这个leader首次写消息，则会在缓冲中增加一个条目；否则就不做更新。而每次副本重新成为leader时会查询这部分缓冲，获取出对应leader版本呢的offset
我们基于同样的情况分析，follower宕机并且恢复之后，有两种情况，如果这个时候leader副本没有挂，也就是意味着没有发生leader选举，那么follower恢复之后并不会去截断自己的日志，而是先发送一个OffsetForleaderEpochRequest请求给到leader副本，leader副本收到请求之后返回当前的LEO。
如果follower副本的leaderEpoch和leader副本 的epoch相同，leader的LEO只可能大于或者等于follower副本的LEO值，所以这个时候不会发生截断。
如果follower副本和leader副本的epoch值不同，那么leader副本会查找follower副本传过来的epoch+1在本地文件中存储的StartOffset返回给follower副本，也就是新leader副本的LEO。这样也避免了数据丢失的问题。
如果leader副本宕机了重新选举新的leader，那么原本的follower副本就会变成leader，意味着epoch从0变成 1，使得原本follower副本中LEO的值得到了保留。
Leader副本的选举过程
1、KafkaConntroller会见你听Zookeeper的/broker/ids节点路径，一旦发现有broker挂了，执行下面的逻辑。这里暂时先不考虑KafkaController所在broker挂了的情况，KafkaController挂了，各个broker会重新选举出新的KafkaController
2、leader副本在该broker上的区分就要重新进行选举leader，目前选举策略是
a) 优先从isr列表中选出第一个作为leader副本，这个叫优先副本，理想情况下有限副本就是该分区的leader副本
b) 如果isr列表为空，则查看该topic的unclean.leader.election.enable配置。
unclean.leader.election.enable：为true则代表允许选用非isr列表的副本作为leader，那么此时就意味着数据可能丢失，为false的话，则表示不允许，直接抛出NoReeplicaOnlineException异常，造成leader副本选举失败。
c) 如果上述配置为true，则从其他副本中选出一个作为leader副本，并且isr列表值包含该leader副本。一旦选举成功，则将选举后的leader和isr和其他副本信息写入到该分区的对应的zk路径上。
消息的存储
消息发送端发送消息到broker上以后，消息是如何持久化的呢？
首先我们需要了解的是，kafka是使用日志文件的方式来保存生产者和发送者的消息，每条消息都有一个offset值来表示它在分区中的偏移量。Kafka中存储的一般都是海量的消息数据，为了避免日志文件过大，Log并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是<topic_name>_<partition_id>
消息的文件存储机制
一个topic的多个partition在物理磁盘上的保存路径，路径保存在/tmp/kafka-logs/topic_partition，包含日志文件、索引文件和时间索引文件。
kafka是通过分段的方式将log分为多个LogSegment，LogSegment是一个逻辑上的概念，一个LogSegment对应磁盘上的一个日志文件和一个索引文件，其中日子文件是用来记录消息的。索引文件是用来保存消息的索引。
LogSegment
假设kafka以partition为最小存储单位，那么我们可以想象当kafka producer不断发送消息，必然回引起partiton文件的无线扩张，这样对于消息文件的维护以及被消费的消息的清理带来非常大的挑战，所以kafka以semgent为单位又把partition进行细分。每个partition相当于一个巨型文件被平均分配到多个大小相等的segment数据文件中(每个segment文件中的消息不一定相等)，这种特性方便已经被消费的消息的清理，提高磁盘的利用率。
1、log.segment.bytes=107370(设置分段大小)，默认是1GB，我们把这个值调小以后，可以看到日志分段的效果。
2、抽取其中三个分段进行分析
segment file由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀“.index"和”.log"分别表示为segment索引文件、数据文件。
segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值进行递增。数值最大为64位long大小，20为数字字符长度，没有数字永0填充。
查看segment文件命名规则
通过下面这条命令可以看到kafka消息日志的内容
sh kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test-
0/00000000000000000000.log --print-data-log
假如第一个log文件的最后一个offset为:5376,所以下一个segment的文件命名为:
00000000000000005376.log。对应的index为00000000000000005376.index
segment中index和log的对应关系
从所有分段中，找一个分段进行分析
为了提高查找消息的性能，为每一个日志文件添加2个索引索引文件：OffsetIndex 和 TimeIndex，分
别对应.index以及.timeindex, TimeIndex索引文件格式：它是映射时间戳和相对offset
查看索引内容：
sh kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test-
0/00000000000000000000.index --print-data-log
如图所示，index中存储了索引以及物理偏移量。 log存储了消息的内容。索引文件的元数据执行对应数据文件中message的物理偏移地址。举个简单的案例来说，以[4053,80899]为例，在log文件中，对应的是第4053条记录，物理偏移量（position）为80899. position是ByteBuffer的指针位置
在partition中如何通过offset查找message
查找的算法是
1、根据offset的值，查找segment段中的index索引文件。由于索引文件命名是以上一个文件的最后一个offset进行命名的，所以，使用二分查找算法能够根据offset快速定位到指定的索引文件。
2、找到索引文件后，根据offset进行定位，找到索引文件中的符号范围的索引。（kafka采用 稀疏索引的方式来提高查找性能）
3、得到position以后，再到对应的log文position处开始查找offset对应的消息，将每条消息的 offset与目标offset进行比较，直到找到消息。
比如说：我们要查找offse=2490这条消息，那么小找到00000000000000000000.index,然后找到[2478，49111]这个索引，再到log文件中，根据4911这个position开始查找，比较每条消息的offset是大于等于2490。最后查找到对应的消息以后返回
Log文件的消息内容分析
前面我们通过kafka提供的命令，可以查看二进制的日志文件信息，一条消息，会包含很多的字段。
offset: 5371 position: 102124 CreateTime: 1531477349286 isvalid: true keysize:
-1 valuesize: 12 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1
sequence: -1 isTransactional: false headerKeys: [] payload: message_5371
offset和position这两个前面已经讲过了、createTime表示创建时间、keysize和valuesize表示key和value的大小、compresscode表示压缩编码、payload：表示消息的具体内容
日志的清除策略以及压缩策略
日志清除策略
前面提到过，日志的分段存储，一方面能够减少单个文件内容的大小，另一方面，方便kafka进行日志清理。日志清理策略由两个：
1、根据消息的保留时间，当消息在kafka中保存的时间超过了指定的时间，就会触发清理过程
2、根据topic存储的数据大小，当topic所占的日志文件大小大于一定的阀值，则可以开始删除最旧的消息。kafka会启动一个后台线程，定期检查是否存在可以删除的消息。
通过log.retention.bytes和log.retention.hours这俩个参数来设置，当其中任意一个达到要求，都会执行删除。
默认保留时间为：7天
磁盘存储的性能问题
磁盘存储的性能优化
我们现在大部分企业仍然用的是机械结构的磁盘，如果把消息以随机的方式写入到磁盘，那么磁盘首先要做的就是寻址，也就是定位到数据所在的物理地址，在磁盘上就要找到对应的柱面、磁头以及对应的扇区；这个过程相对内存来说会消耗大量时间，为了规避随机读写带来的时间消耗，kafka采用顺序写的方式存储数据。即使是这样，但是频繁的I/O操作仍然会造成磁盘的性能瓶颈
零拷贝
消息从发送到落地保存，broker维护的消息日志本身就是文件目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。在消费者获取消息时，服务器先从硬盘读取数据到内存，然后把内存中的数据原封不动的通过socket发送给消费者。虽然这个操作描述起来很简单，但实际上经历了很多步骤。
操作系统将数据从磁盘读入到内核空间的页缓存
▪ 应用程序将数据从内核空间读入到用户空间缓存中
▪ 应用程序将数据写回到内核空间到socket缓存中
▪ 操作系统将数据从socket缓冲区复制到网卡缓冲区，以便将数据经网络发出
通过“零拷贝”技术，可以去掉这些没必要的数据复制操作，同时也会减少上下文切换次数。现代的unix操作系统提供一个优化的代码路径，用于将数据从页缓存传输到socket；在Linux中，是通过sendfile系统调用来完成的。Java提供了访问这个系统调用的方法：FileChannel.transferTo API使用sendfile，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的
页缓存
页缓存是操作系统实现的一种主要的磁盘缓存，但凡设计到缓存的，基本都是为了提升i/o性能，所以页缓存是用来减少磁盘I/O操作的。
磁盘高速缓存有两个重要因素：
第一，访问磁盘的速度要远低于访问内存的速度，若从处理器L1和L2高速缓存访问则速度更快。
第二，数据一旦被访问，就很有可能短时间内再次访问。正是由于基于访问内存比磁盘快的多，所以磁盘的内存缓存将给系统存储性能带来质的飞越。当 一 个进程准备读取磁盘上的文件内容时， 操作系统会先查看待读取的数据所在的页(page)是否在页缓存(pagecache)中，如果存在（命中）则直接返回数据， 从而避免了对物理磁盘的I/0操作；如果没有
命中， 则操作系统会向磁盘发起读取请求并将读取的数据页存入页缓存， 之后再将数据返回给进程。同样，如果 一 个进程需要将数据写入磁盘， 那么操作系统也会检测数据对应的页是否在页缓存中，如果不存在， 则会先在页缓存中添加相应的页， 最后将数据写入对应的页。 被修改过后的页也就变成了脏页， 操作系统会在合适的时间把脏页中的数据写入磁盘， 以保持数据的 一 致性Kafka中大量使用了页缓存， 这是Kafka实现高吞吐的重要因素之 一 。 虽然消息都是先被写入页缓存，然后由操作系统负责具体的刷盘任务的， 但在Kafka中同样提供了同步刷盘及间断性强制刷盘(fsync),可以通过 log.flush.interval.messages 和log.flush.interval.ms 参数来控制。同步刷盘能够保证消息的可靠性，避免因为宕机导致页缓存数据还未完成同步时造成的数据丢失。但是实际使用上，我们没必要去考虑这样的因素以及这种问题带来的损失，消息可靠性可以由多副本来解决，同步刷盘会带来性能的影响。 刷盘的操作由操作系统去完成即可
Kafka消息的可靠性
没有一个中间件能够做到百分百的完全可靠性，可靠性更多的还是基于几个9的衡量指标，比如4个9、5个9。软件系统的可靠性只能够无限去接近100%，但不可能达到100%。所以kafka如何实现最大可能的可靠性呢？
1、分区副本，你可以创建更多的分区提高可靠性，但是分区数过多也会带来性能上的开销，一般来说，3个副本就能满足对大部分场景的可靠性要求。
2、acks，生产者发送消息的可靠性，也就是我要保证我这个消息一定是到了broker并且完成了多副本的持久化，但这种要求也同样会带来性能上的开销。它有几个可选项
2.1、1：生产者把消息发送到leader副本，leader副本在成写入到本地日志之后 就告诉生产者消息提交成功，但是如果isr集合中的follower副本还没来得及同步leader副本的消息，leader挂了，就会造成消息丢失。
2.2、-1：消息不仅仅写入到leader副本，并且ISR集合中所有副本同步 完成之后才告诉生产者已经提交成功，这个时候即使leader副本挂了也不会造成数据丢失。
2.3、0：表示producer不需要等待broker的消息确认。这个选项时延最小但同时风险最大(因为达能server宕机时，数据将会丢失)
2、保障消息到了broker之后，消费者也需要有一定的保证，因为消费者也可能出现某些问题导致消息没有消费到
3、enable.auto.commit默认为true，也就是自动提交offset，自动他提交是批量执行的 ，有一个时间窗口，这种方式会带来重复提交或消息丢失问题，所以对于高可靠性要求的程序，要使用手动提交。对于高可靠要求的应用来说，宁愿重复消费也不应该因为消费异常而导致消息丢失