Kafka

Kafka概要

Kafka简介

Kafka官网上将其描述为一个分布式流平台，Kafka以一种全新的视角，将其处理的数据看做流。那么首先看下一个流平台需要有什么核心能力呢：
（1）可以发布和订阅记录流，类似消息队列和企业级消息系统
（2）能够持久存储记录流，并具有容错性
（3）可实时处理记录流
上面已经提到了Kafka将自己定位为一个分布式流平台，因此在设计上就包含了分布式流平台的能力，主要包含以下几点：

• 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能
• 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒10W条消息的传输
• 支持Kafka服务器间的消息分区，分布式消费，同时保证每个partition内的消息顺序传输
• 同时支持离线数据处理和实时数据处理

Kafka的几个概念

Topic
消息的主题，用来标识一类消息，是一个逻辑概念
Partition
分区，一个主题的消息会按照一定的规则进行分区，满足同一规则的消息放在同一个分区上进行存储

每个分区对应一个log文件，而log文件又被分为多个segment进行存储，每个segment对应一个log文件（可配置其有效时间和最大容量）和index文件，index文件记录存储数据对应的偏移量，log文件存储生产者生产的数据。类似如下的结构：

00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log

下面演示了查找定位一个消息的过程：

分区的好处：（1）方便在集群中扩展（2）可以提高并发（读写在一个partition）
分区的方法：（1）指定分区（2）使用key进行hash取余（3）随机值进行hash取余
Broker
每一台Kafka服务器都是一个Broker，多个Broker构成一个集群。每个集群都有一个 broker 同时充当了集群控制器的角色(自动从集群的活跃成员中选举出来)。控制器负责管理工作，包括将分区分配给 broker和监控 broker. 在集群中， 一个分区从属于一个 broker, 该broker被称为分区的Leader。一个分区可以分配给多个broker，这个时候会发生分区复制，这些复制出来的分区成为Follower。这种复制机制为分区提供了消息冗余，如果有一个 broker失效，其他 broker可以接管领导权。不过，相关的消费者和生产者都要重新连接到新的Leader（读和写都在Leader上进行的）
副本：每个broker上的leader的数据再其他broker上都有N份副本来保证数据的可靠性，副本的数据同步策略主要有以下两种：kafka选择了第二种，并在第二种基础上进行了优化，使用ISR（in-sync replica set）意为和leader同步的集合，如果follwer长期未向leader同步数据，就会将其踢出（时间由参数指定）

Producer
消息生产者，向broker中push消息
Consumer
消息消费者，从broker中pull消息
Comsumer Group
由过个消费者组成的一个组，一条消息只会被组里的一个消费者消费

Kafka的几个保证

（1）所有生产者生产的消息，都按照生产的先后顺序进行保存
（2）消费者看到的消息顺序就是它的存放顺序
（3）对于有N个拷贝的主题，可以允许最多N-1个服务丢失已经提交日志的记录
第3点为什么个数是N-1是由其选举策略来决定的，而对于Zookeeper，如果从服务器有2N+1个，那么最多允许N个失败，因为其选择策略是少数服务多数，需至少N+1个从服务器的投票结果

Kafka设计思想

Kafka架构

Kafka原理剖析

Push&Pull
Kafka采用Push和Pull模式，Producer向Broker中Push消息，Consumer从Broker中读取消息
生产Partition机制
如何平衡Partition：对消息的Key进行hash来决定消息所在的Partition

消费机制
提供的消费能力
一个消费组只会消费一个消息一次，但可以被多个消费组消费

如何保证一个消费组一条消息只消费一次？
通过保证只有一个Consumer会消费某条消息，实际上是通过partition只会被分配到一个Consumer上
这样设计的劣势是无法保证同一个Consumer Group里的Consumer均匀消费数据，优势是每个Consumer不用都跟大量的Broker通信，减少通信开销，同时也降低了分配难度，实现也更简单。另外，因为同一个Partition里的数据是有序的，这种设计可以保证每个Partition里的数据可以被有序消费。
如果某Consumer Group中Consumer（每个Consumer只创建1个MessageStream）数量少于Partition数量，则至少有一个Consumer会消费多个Partition的数据，如果Consumer的数量与Partition数量相同，则正好一个Consumer消费一个Partition的数据。而如果Consumer的数量多于Partition的数量时，会有部分Consumer无法消费该Topic下任何一条消息。
如何有序消费？
因为消息是在每个分区顺序写入的，因此可以保证在分区内是有序消费的
可以保证全局有序么？
不能保证全局有序。 是否有保证机制？

Broker存储机制
一个partition的数据只会被一个消费者消费，一个消费者可能会消费多个partition，
broker的变更和消费者数量的变更都会触发消费者-Partition关系的rebalance

常用的数据复制和一致性方案
Master-Slave
同步复制
保证强一致性而降低可用性
异步复制
保证可用性而降低一致性

Paxos及其变种方案
Paxos、Zab、RAFT等

基于ISR的数据复制方案
类似Master-Slave方式，但既不是完全是同步复制，也不完全是异步复制，每个Leader都维护一个ISR（In-Sync-Replica），这个ISR列表包含了所有与自己同步的follow列表，包括自己本身。每次数据写入时，只有ISR中的所有被复制完，才可以将数据设置为Commit，才能被消费者消费。但这个过程不是阻塞的，在等待Follwer的同时，leader可以接受新的消息

ISR是动态维护的，如果某个Follwer不能跟上，则会被移除（移除的标准，按照多长时间没有发起同步请求，可以按照落后的条数来判断），如果跟上则会再被放入，每次ISR发生变更，都会在Zookeeper中进行持久化。

使用ISR的好处：
（1）避免最慢的Follower拖慢整个的速度
（2）由于只有Commit过的消息才能被消费者消费，因此对于消费者而言，所有ISR中的数据都处于同步状态，相对于异步同步，提高了一致性
（3）ISR可以动态调整，极限情况下，可以只包含Leader，相对于多数同步方案，容忍相同失败的节点个数，需要总节点个数减少了近一半

高性能实现

利用partition并行处理

不同partition位于不同的机器，充分利用集群的优势，实现并行处理

顺序写磁盘

每个Partition由过个Segment来组成，每个Segment对应一个物理文件，当有新的数据，采用在文件后面直接追加的方式，当需要进行数据清理时，会将整个文件删除。顺序写磁盘减少了随机读写带来的性能消耗

充分利用Page cache**

引入 Cache 层的目的是为了提高 Linux 操作系统对磁盘访问的性能。Cache 层在内存中缓存了磁盘上的部分数据。当数据的请求到达时，如果在 Cache 中存在该数据且是最新的，则直接将数据传递给用户程序，免除了对底层磁盘的操作，提高了性能。Cache 层也正是磁盘 IOPS 为什么能突破 200 的主要原因之一。
在 Linux 的实现中，文件 Cache 分为两个层面，一是 Page Cache，另一个 Buffer Cache，每一个 Page Cache 包含若干 Buffer Cache。Page Cache 主要用来作为文件系统上的文件数据的缓存来用，尤其是针对当进程对文件有 read/write 操作的时候。Buffer Cache 则主要是设计用来在系统对块设备进行读写的时候，对块进行数据缓存的系统来使用。
Broker 收到数据后，写磁盘时只是将数据写入 Page Cache，并不保证数据一定完全写入磁盘。从这一点看，可能会造成机器宕机时，Page Cache 内的数据未写入磁盘从而造成数据丢失。但是这种丢失只发生在机器断电等造成操作系统不工作的场景，而这种场景完全可以由 Kafka 层面的 Replication 机制去解决。如果为了保证这种情况下数据不丢失而强制将 Page Cache 中的数据 Flush 到磁盘，反而会降低性能。也正因如此，Kafka 虽然提供了 flush.messages 和 flush.ms 两个参数将 Page Cache 中的数据强制 Flush 到磁盘，但是 Kafka 并不建议使用。

零拷贝

拷贝主要发生在以下两种场景：

1. 从网络拷贝到磁盘（写数据，Producer到Broker）
2. 从磁盘拷贝到网络（读数据，Broker到Comsumer）

传统拷贝
这一过程实际上发生了四次数据拷贝。首先通过系统调用将文件数据读入到内核态Buffer（DMA拷贝），然后应用程序将内存态Buffer数据读入到用户态Buffer（CPU拷贝），接着用户程序通过Socket发送数据时将用户态Buffer数据拷贝到内核态Buffer（CPU拷贝），最后通过DMA拷贝将数据拷贝到NIC Buffer。同时，还伴随着四次上下文切换，如下图所示。

传统的读写都需要经过四次的拷贝：两次cpu拷贝和两次DMA拷贝，以及内核空间到用户空间的上下文切换

零拷贝

减少网络开销

批处理
Kafka客户端和Broker在进行网络发送之前，会累积多条记录（读或写）
数据压缩
对传输的数据进行压缩，支持的有gzip、snappy、lz4

Kafka参数配置

生产者
生产者可以包含三种发送场景，发送并忘记、同步发送、异步发送
生产者可以设置ack参数，来表示有多少个分区副本收到消息，才认为发送成功
ack=0
生产者在成功写入悄息之前不会等待任何来自服务器的响应。这种会存在数据丢失
ack=1
只要集群的首领节点收到消息，生产者就会收到 一个来自服务器的成功响应。这种如果在follwer同步之前，leader挂了会丢失数据，因此具有At Most Once（最多一次）语义
ack=-1
只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自 服务器的成功响应。这种如果follwer同步完之后，broker发送ack之前，leader挂了会存在数据重复，因此具有At Least Once（至少一次）语义


故障处理
LEO（Log End Offset）：每个副本的最后一个offset
HW（High Water）：值得是消费者能见到的最大offset，ISR队列中的最小的LEO，消费者与Leader通信时，只能消费HW之前的数据，之后的数据不可见
当Leader或Follower发生故障后，高于HW的数据会被截掉，因此数据的一致性不能保证数据的不丢失或不重复，这个是由ack来控制的，HW只保证数据的一致性。

消费者
通过增加一个消费组里的消费者来提高消费的效率，但是同一个消费组里消费者的个数不要超过Topic的分组个数。Kafka还支持消息被多个消费组同时消费

Kafka应用场景

消息系统

这种主要用于常见的发布/订阅模式。这个领域相关的还有RabbitMQ、RocketMQ等
生产者生产一个消息，所有消费者消费这个消息进行其他处理，比如平台入口接口生成一个订单消息，各个具体业务进行订阅，进行各自业务处理

日志收集

这种主要用于收集服务器上的日志。
对于一个上百台的集群，如何快速定位问题是一个考验，可以通过Kafka进行日志收集，并将结果写入ES等，便于快速检索

存储系统

流处理系统

这种主要用于需要进行实时计算的场景。
比如通过Spark Streaming进行实时计算的场景

活动跟踪

用于记录一个用户在一个平台上的活动路径，可以为用户的每一种操作定义一种主题，有此来统计和分析用户的行为，为机器学习和推荐提供数据

参考文档

1. http://www.jasongj.com/tags/Kafka/
2. https://mp.weixin.qq.com/s/dbnpPEF0FBB5A5xH21OoeQ
3. https://mp.weixin.qq.com/s/ITLN-DHxYc5w6qrlFD8HWQ