大数据求索：Kafka的重要原理和概念（1）

一、Kakfa简介

Apache kafka是一个分布式的基于push-subscribe的消息系统，它具备快速、可扩展、可持久化的特点。它现在是Apache旗下的一个开源系统，作为hadoop生态系统的一部分，被各种商业公司广泛应用。它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/spark流式处理引擎。

二、Kafka技术概览

2.1 Kafka的特性

1）高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。

2）可扩展性：kafka集群支持热扩展

3）持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

4）容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

5）高并发：支持数千个客户端同时读写。

2.2 Kafka的重要设计思想

Topic & Partition：

Topic相当于传统消息系统MQ中的一个队列queue，producer端发送的message必须指定是发送到哪个topic，但是不需要指定topic下的哪个partition，因为kafka会把收到的message进行load balance，均匀的分布在这个topic下的不同的partition上（ hash(message) % [broker数量] ）。

物理上存储上，这个topic会分成一个或多个partition，每个partiton相当于是一个子queue。在物理结构上，每个partition对应一个物理的目录（文件夹），文件夹命名是[topicname][partition][序号]，一个topic可以有无数多的partition，根据业务需求和数据量来设置。在kafka配置文件中可随时更改num.partitions参数来配置更改topic的partition数量，也可以在创建Topic时通过参数指定parittion数量。Topic创建之后通过Kafka提供的工具也可以修改partiton数量。

一般来说，（1）一个topic的partition数量大于等于Broker的数量，可以提高吞吐率。（2）同一个partition的replication尽量分散到不同的机器，高可用。

当行增加一个partition的时候，partition里面的message不会重新进行分配，原来的partition里面的message数据不会变，新加的这个partition刚开始是空的，随后进入这个topic的message就会重新参与所有partition的load balance。

partition replication：

每个partition可以在其他的kafka broker节点上存副本，以便某个kafka broker节点宕机不会影响这个kafka集群。存replica副本的方式是按照kafka broker的顺序存。例如有5个kafka broker节点，某个topic有3个partition，每个partition存2个副本，那么partition1存broker1,broker2，partition2存broker2,broker3...以此类推（replication副本数目不能大于kafka broker节点的数目，否则报错。这里的replica数其实就是partition的副本总数，其中包括一个leader，其他的就是copy副本）。这样如果某个broker宕机，其实整个kafka内数据依然是完整的。但是，replica副本数越高，系统虽然越稳定，但是回来带资源和性能上的下降；replica副本少的话，也会造成系统丢数据的风险。

**Partition leader与follower：

**partition也有leader和follower之分。leader是主partition，producer写kafka的时候先写partition leader，再由partition leader push给其他的partition follower。partition leader与follower的信息受Zookeeper控制，一旦partition leader所在的broker节点宕机，zookeeper会从其他的broker的partition follower上选择follower变为parition leader。

Topic分配partition和partition replica的算法：

（1）将Broker（size=n）和待分配的Partition排序。

（2）将第i个Partition分配到第（i%n）个Broker上。

（3）将第i个Partition的第j个Replica分配到第（(i + j) % n）个Broker上

Kakfa Broker Leader的选举：

Kakfa Broker集群受Zookeeper管理，所以Kafka的使用需要先安装Zookeeper。所有的Kafka Broker节点一起去Zookeeper上注册一个临时节点。但是只有一个Kafka Broker会注册成功，其他的都会失败。这个成功在Zookeeper上注册临时节点Kafka Broker会成为Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower（这个过程叫Controller在ZooKeeper注册Watch）。

注册成功后 ，Controller会监听其他的Kafka Broker的所有信息，如果这个Controller宕机了，在zookeeper上面的那个临时节点就会消失。此时，所有的kafka broker又会一起去zookeeper上注册一个临时节点。例如：一旦有一个broker宕机了，这个kafka broker controller会读取该宕机broker上所有的partition在zookeeper上的状态，并选取ISR列表中的一个replication作为partition leader（如果ISR列表中的replication全挂，选一个幸存的replication作为leader; 如果该partition的所有的replication都宕机了，则将新的leader设置为-1，等待恢复，等待ISR中的任一个replication“活”过来，并且选它作为leader；或选择第一个“活”过来的replication（不一定是ISR中的）作为leader）

Consumergroup:

各个consumer（consumer 线程）可以组成一个组（Consumer group ），partition中的每个message只能被组中的一个consumer消费，如果一个message可以被多个consumer消费的话，那么这些consumer必须在不同的组。所以如果想同时对一个topic做消费的话，启动多个consumer group就可以了，但是要注意的是，这里的多个consumer的消费都必须是顺序读取partition里面的message，新启动的consumer默认从partition队列最头端最新的地方开始阻塞的读message。它不能像AMQ那样可以多个BET作为consumer去互斥的（for update悲观锁）并发处理message，这是因为多个consumer去消费一个Queue中的数据的时候，由于要保证不能多个线程拿同一条message，所以就需要行级别悲观锁（for update），这就导致了consume的性能下降，吞吐量不够。

而kafka为了保证吞吐量，只允许同一个consumer group下的一个consumer线程去访问一个partition。如果觉得效率不高的时候，可以加partition的数量来横向扩展，在加新的consumer thread中去消费。如果想多个不同的业务都需要消费这个topic的数据，起多个consumer group就好了，大家都是顺序的读取message，offsite的值互不影响。这样没有锁竞争，充分发挥了横向的扩展性，吞吐量极高。这也就形成了分布式消费的概念。

当启动一个consumer group去消费一个topic的时候，无论topic里面有多个少个partition，无论我们consumer group里面配置了多少个consumer thread，这个consumer group下面的所有consumer thread一定会消费全部的partition，即便这个consumer group下只有一个consumer thread，那么这个consumer thread也会去消费所有的partition。因此，最优的设计就是，consumer group下的consumer thread的数量等于partition数量，这样效率是最高的。

消息状态：

在Kafka中，消息的状态被保存在consumer中，broker不会关心哪个消息被消费了被谁消费了，只记录一个offset值（指向partition中下一个要被消费的消息位置），这就意味着如果consumer处理不好的话，broker上的一个消息可能会被消费多次。

消息持久化：

Kafka中会把消息持久化到本地文件系统中，并且保持极高的效率。

消息有效期：

Kafka会长久保留其中的消息，以便consumer可以多次消费，当然其中很多细节是可配置的。

批量发送：

Kafka支持以消息集合为单位进行批量发送，以提高push效率。

Consumer：

Consumer处理partition里面的message的时候是O(1)顺序读取的，所以必须维护着上一次读到哪里的offsite信息。high level API的offset存于Zookeeper中，low level API的offset由自己维护。一般来说都是使用high level api。Consumer的delivery gurarantee，默认是读完message先commmit再处理message，autocommit默认是true，这时候先commit就会更新offsite+1，一旦处理失败，offsite已经+1，这个时候就会丢message；也可以配置成读完消息处理再commit，这种情况下consumer端的响应就会比较慢的，需要等处理完才行。

一般情况下，一定是一个consumer group处理一个topic的message。Best Practice是这个consumer group里面consumer的数量等于topic里面partition的数量，这样效率是最高的，一个consumer thread处理一个partition。如果这个consumer group里面consumer的数量小于topic里面partition的数量，就会有consumer thread同时处理多个partition（这个是kafka自动的机制，我们不用指定），但是总之这个topic里面的所有partition都会被处理到的。如果这个consumer group里面consumer的数量大于topic里面partition的数量，多出的consumer thread就会闲着啥也不干，剩下的是一个consumer thread处理一个partition，这就造成了资源的浪费，因为一个partition不可能同时被两个consumer thread去处理。

同步异步：

Producer采用异步push方式，极大提高Kafka系统的吞吐率（可以通过参数控制是采用同步还是异步方式）

离线数据装载：

Kafka由于对可拓展的数据持久化的支持，它也非常适合向Hadoop或者数据仓库中进行数据装载。

push-and-pull :

Kafka中的Producer和consumer采用的是push-and-pull模式，即Producer只管向broker push消息，consumer只管从broker pull消息，两者对消息的生产和消费是异步的。

三、Kafka核心特性

3.1 消息可靠性

在消息系统中，保证消息在生产和消费过程中的可靠性是十分重要的，在实际消息传递过程中，可能会出现如下三种情况：

一个消息发送失败

一个消息被发送多次

最理想的情况：exactly-once ,一个消息发送成功且仅发送了一次

一个消息如何算投递成功，Kafka提供了三种模式：

第一种是啥都不管，发送出去就当作成功，这种情况当然不能保证消息成功投递到broker；

第二种是Master-Slave模型，只有当Master和所有Slave都接收到消息时，才算投递成功，这种模型提供了最高的投递可靠性，但是损伤了性能；

第三种模型，即只要Master确认收到消息就算投递成功；实际使用时，根据应用特性选择，绝大多数情况下都会中和可靠性和性能选择第三种模型。

消息在broker上的可靠性，因为消息会持久化到磁盘上，所以如果正常stop一个broker，其上的数据不会丢失；但是如果不正常stop，可能会使存在页面缓存来不及写入磁盘的消息丢失，这可以通过配置flush页面缓存的周期、阈值缓解，但是同样会频繁的写磁盘会影响性能，又是一个选择题，根据实际情况配置。

消息消费的可靠性，Kafka提供的是**“At least once”** 模型，因为消息的读取进度由offset提供，offset可以由消费者自己维护也可以维护在zookeeper里，但是当消息消费后consumer挂掉，offset没有即时写回，就有可能发生重复读的情况，这种情况同样可以通过调整commit offset周期、阈值缓解，甚至消费者自己把消费和commit offset做成一个事务解决，但是如果你的应用不在乎重复消费，那就干脆不要解决，以换取最大的性能。

可以通过ack参数控制。

当ack=0，表示producer不会等待broker的响应，所以，producer无法知道消息是否发送成功，这样有可能会导致数据丢失，但同时，acks值为0会得到最大的系统吞吐量。

当ack=1，表示producer会在leader partition收到消息时得到broker的一个确认，这样会有更好的可靠性，因为客户端会等待直到broker确认收到消息。

当ack=-1，producer会在所有备份的partition收到消息时得到broker的确认，这个设置可以得到最高的可靠性保证。

从Producer端看：Kafka是这么处理的，当一个消息被发送后，Producer会等待broker成功接收到消息的反馈（可通过参数控制等待时间），如果消息在途中丢失或是其中一个broker挂掉，Producer会重新发送(我们知道Kafka有备份机制，可以通过参数控制是否等待所有备份节点都收到消息)。

从Consumer端看：前面讲到过partition，broker端记录了partition中的一个offset值，这个值指向Consumer下一个即将消费message。当Consumer收到了消息，但却在处理过程中挂掉，此时Consumer可以通过这个offset值重新找到上一个消息再进行处理。Consumer还有权限控制这个offset值，对持久化到broker端的消息做任意处理。

3.2 压缩

Kafka支持以集合（batch）为单位发送消息，在此基础上，Kafka还支持对消息集合进行压缩。Producer端可以通过GZIP或Snappy格式对消息集合进行压缩。Producer端进行压缩之后，在Consumer端需进行解压 。压缩的好处就是减少传输的数据量，减轻对网络传输的压力，在对大数据处理上，瓶颈往往体现在网络上而不是CPU（压缩和解压会耗掉部分CPU资源）。

那么如何区分消息是压缩的还是未压缩的呢，Kafka在消息头部添加了一个描述压缩属性字节，这个字节的后两位表示消息的压缩采用的编码，如果后两位为0，则表示消息未被压缩。

3.3 备份机制

备份机制是Kafka0.8版本 的新特性，备份机制的出现大大提高了Kafka集群的可靠性、稳定性。有了备份机制后，Kafka允许集群中的节点挂掉后而不影响整个集群工作。**一个备份数量为n的集群允许n-1个节点失败。**在所有备份节点中，有一个节点作为lead节点，这个节点保存了其它备份节点列表，并维持各个备份间的状体同步。下面这幅图解释了Kafka的备份机制:

3.4 无状态的Kafka Broker

broker没有副本机制,一旦broker宕机,该broker的消息将都不可用；

broker不保存订阅者的状态,由订阅者自己保存；

无状态导致消息的删除成为难题（可能删除的消息正在被订阅），kafka采用基于时间的SLA(服务水平保证)，消息保存一定时间（通常为7天）后会被删除。

消息订阅者可以rewind back到任意位置重新进行消费，当订阅者故障时，可以选择最小的offset进行重新读取消费消息。

3.5 message的交付与生命周期

不是严格的JMS， 因此kafka对消息的重复、丢失、错误以及顺序型没有严格的要求。（这是与AMQ最大的区别）

kafka提供at-least-once delivery,即当consumer宕机后，有些消息可能会被重复delivery。

因每个partition只会被consumer group内的一个consumer消费，故kafka保证每个partition内的消息会被顺序的订阅。

Kafka为每条消息为每条消息计算CRC校验，用于错误检测，CRC校验不通过的消息会直接被丢弃掉。