• group.id: A unique string that identifies the consumer group this consumer belongs to.

• client.id: The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request.

• bootstrap.servers: A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.

• key.deserializer: Deserializer class for key that implements the org.apache.kafka.common.serialization.Deserializer interface.

• value.deserializer: Deserializer class for value that implements the org.apache.kafka.common.serialization.Deserializer interface.

• fetch.min.bytes: The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request.

• fetch.max.bytes: The maximum amount of data the server should return for a fetch request.

• max.partition.fetch.bytes: The maximum amount of data per-partition the server will return.

• max.poll.records: The maximum number of records returned in a single call to poll().

• heartbeat.interval.ms: The expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities.

• session.timeout.ms: The timeout used to detect consumer failures when using Kafka’s group management facility.

• enable.auto.commit: If true the consumer’s offset will be periodically committed in the background.

ISR: Kafka在ZK中动态维护了一个ISR(In-Sync Replica)，即保持同步的副本列表，该列表中保存的是与leader副本保持消息同步的所有副本对应的brokerId。如果一个副本宕机或者落后太多，则该follower副本将从ISR列表中移除。

Zookeeper:

• Kafka利用ZK保存相应的元数据信息，包括：broker信息，Kafka集群信息，旧版消费者信息以及消费偏移量信息，主题信息，分区状态信息，分区副本分片方案信息，动态配置信息，等等。

• Kafka在zk中注册节点说明：

  • /consumers: 旧版消费者启动后会在ZK的该节点下创建一个消费者的节点

  • /brokers/seqid: 辅助生成的brokerId，当用户没有配置broker.id时，ZK会自动生成一个全局唯一的id。

  • /brokers/topics: 每创建一个主题就会在该目录下创建一个与该主题同名的节点。

  • /borkers/ids: 当Kafka每启动一个KafkaServer时就会在该目录下创建一个名为{broker.id}的子节点

  • /config/topics: 存储动态修改主题级别的配置信息

  • /config/clients: 存储动态修改客户端级别的配置信息

  • /config/changes: 动态修改配置时存储相应的信息

  • /admin/delete_topics: 在对主题进行删除操作时保存待删除主题的信息

  • /cluster/id: 保存集群id信息

  • /controller: 保存控制器对应的brokerId信息等

  • /isr_change_notification: 保存Kafka副本ISR列表发生变化时通知的相应路径

• Kafka在启动或者运行过程中会在ZK上创建相应的节点来保存元数据信息，通过监听机制在这些节点注册相应的监听器来监听节点元数据的变化。

TIPS

如果跟ES对应，Broker相当于Node，Topic相当于Index，Message相对于Document，而Partition相当于shard。LogSegment相对于ES的Segment。

如何查看消息内容（Dump Log Segments）

 
我们在使用kafka的过程中有时候可以需要查看我们生产的消息的各种信息，这些消息是存储在kafka的日志文件中的。由于日志文件的特殊格式，我们是无法直接查看日志文件中的信息内容。Kafka提供了一个命令，可以将二进制分段日志文件转储为字符类型的文件：

 
$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments	
Parse a log file and dump its contents to the console, useful for debugging a seemingly corrupt log segment.	
Option                                  Description                           	
------                                  -----------                           	
--deep-iteration                        使用深迭代而不是浅迭代                          	
--files <file1, file2, ...>             必填。输入的日志段文件，逗号分隔	
--key-decoder-class                     自定义key值反序列化器。必须实现`kafka.serializer.Decoder` trait。所在jar包需要放在`kafka/libs`目录下。（默认是`kafka.serializer.StringDecoder`）。	
--max-message-size <Integer: size>      消息最大的字节数(默认为5242880)                           	
--print-data-log                        同时打印出日志消息             	
--value-decoder-class                   自定义value值反序列化器。必须实现`kafka.serializer.Decoder` trait。所在jar包需要放在`kafka/libs`目录下。（默认是`kafka.serializer.StringDecoder`）。	
--verify-index-only                     只是验证索引不打印索引内容
 
（滑动查看）

 
 


 
$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test-0/00000000000000000000.log --print-data-log 	
Dumping /tmp/kafka-logs/test-0/00000000000000000000.log	
Starting offset: 0	
offset: 0 position: 0 CreateTime: 1498104812192 isvalid: true payloadsize: 11 magic: 1 compresscodec: NONE crc: 3271928089 payload: hello world	
offset: 1 position: 45 CreateTime: 1498104813269 isvalid: true payloadsize: 14 magic: 1 compresscodec: NONE crc: 242183772 payload: hello everyone
 
（滑动查看）
 


 
注意：这里 --print-data-log  是表示查看消息内容的，不加此项只能看到Header，看不到payload。
 


 
 
也可以用来查看index文件：

 
$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test-0/00000000000000000000.index  --print-data-log 	
Dumping /tmp/kafka-logs/test-0/00000000000000000000.index	
offset: 0 position: 0
 
（滑动查看）
 


 
timeindex文件也是OK的：
 
 
$ bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test-0/00000000000000000000.timeindex  --print-data-log 	
Dumping /tmp/kafka-logs/test-0/00000000000000000000.timeindex	
timestamp: 1498104813269 offset: 1	
Found timestamp mismatch in :/tmp/kafka-logs/test-0/00000000000000000000.timeindex	
  Index timestamp: 0, log timestamp: 1498104812192	
Found out of order timestamp in :/tmp/kafka-logs/test-0/00000000000000000000.timeindex	
  Index timestamp: 0, Previously indexed timestamp: 1498104813269
 
（滑动查看）
 


 
消费者平衡过程
 


 
消费者平衡(Consumer Rebalance)是指的是消费者重新加入消费组，并重新分配分区给消费者的过程。在以下情况下会引起消费者平衡操作:
 
新的消费者加入消费组
当前消费者从消费组退出（不管是异常退出还是正常关闭）
消费者取消对某个主题的订阅
订阅主题的分区增加(Kafka的分区数可以动态增加但是不能减少)
broker宕机新的协调器当选
当消费者在${session.timeout.ms}时间内还没有发送心跳请求，组协调器认为消费者已退出。


 
消费者自动平衡操作提供了消费者的高可用和高可扩展性，这样当我们增加或者减少消费者或者分区数的时候，不需要关心底层消费者和分区的分配关系。但是需要注意的是，在rebalancing过程中，由于需要给消费者重新分配分区，所以会出现在一个短暂时间内消费者不能拉取消息的状况。
 


 
NOTES
 
这里要特别注意最后一种情况，就是所谓的慢消费者(Slow Consumers)。如果没有在session.timeout.ms时间内收到心跳请求，协调者可以将慢消费者从组中移除。通常，如果消息处理比session.timeout.ms慢，就会成为慢消费者。导致两次poll()方法的调用间隔比session.timeout.ms时间长。由于心跳只在 poll()调用时才会发送（在0.10.1.0版本中, 客户端心跳在后台异步发送了），这就会导致协调者标记慢消费者死亡。
 


 
如果没有在session.timeout.ms时间内收到心跳请求，协调者标记消费者死亡并且断开和它的连接。同时，通过向组内其他消费者的HeartbeatResponse中发送IllegalGeneration错误代码 触发rebalance操作。
 
在手动commit offset的模式下，要特别注意这个问题，否则会出现commit不上的情况。导致一直在重复消费。
 


 
二、Kafka的特点
 


 
消息顺序：保证每个partition内部的顺序，但是不保证跨partition的全局顺序。如果需要全局消息有序，topic只能有一个partition。
consumer group：consumer group中的consumer并发获取消息，但是为了保证partition消息的顺序性，每个partition只会由一个consumer消费。因此consumer group中的consumer数量需要小于等于topic的partition个数。（如需全局消息有序，只能有一个partition，一个consumer）
同一Topic的一条消息只能被同一个Consumer Group内的一个Consumer消费，但多个Consumer Group可同时消费这一消息。这是Kafka用来实现一个Topic消息的广播（发给所有的Consumer）和单播（发给某一个Consumer）的手段。一个Topic可以对应多个Consumer Group。如果需要实现广播，只要每个Consumer有一个独立的Group就可以了。要实现单播只要所有的Consumer在同一个Group里。
Producer Push消息，Client Pull消息模式：一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume，采用push模式。事实上，push模式和pull模式各有优劣。push模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。push模式的目标是尽可能以最快速度传递消息，但是这样很容易造成Consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据Consumer的消费能力以适当的速率消费消息。pull模式可简化broker的设计，Consumer可自主控制消费消息的速率，同时Consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。


 
实际上，Kafka的设计理念之一就是同时提供离线处理和实时处理。根据这一特性，可以使用Storm或Spark Streaming这种实时流处理系统对消息进行实时在线处理，同时使用Hadoop这种批处理系统进行离线处理，还可以同时将数据实时备份到另一个数据中心，只需要保证这三个操作所使用的Consumer属于不同的Consumer Group即可。
 


 
三、kafka的HA
 


 
Kafka在0.8以前的版本中，并不提供High Availablity机制，一旦一个或多个Broker宕机，则宕机期间其上所有Partition都无法继续提供服务。若该Broker永远不能再恢复，亦或磁盘故障，则其上数据将丢失。而Kafka的设计目标之一即是提供数据持久化，同时对于分布式系统来说，尤其当集群规模上升到一定程度后，一台或者多台机器宕机的可能性大大提高，对Failover要求非常高。因此，Kafka从0.8开始提供High Availability机制。主要表现在Data Replication和Leader Election两方面。

 


 
Data Replication
 


 
 
Kafka从0.8开始提供partition级别的replication，replication的数量可在
 
$KAFKA_HOME/config/server.properties 中配置:

 
default.replication.factor = 1
 


 
该 Replication与leader election配合提供了自动的failover机制。replication对Kafka的吞吐率是有一定影响的，但极大的增强了可用性。默认情况下，Kafka的replication数量为1。每个partition都有一个唯一的leader，所有的读写操作都在leader上完成，follower批量从leader上pull数据。一般情况下partition的数量大于等于broker的数量，并且所有partition的leader均匀分布在broker上。follower上的日志和其leader上的完全一样。
 


 
需要注意的是，replication factor并不会影响consumer的吞吐率测试，因为consumer只会从每个partition的leader读数据，而与replicaiton factor无关。同样，consumer吞吐率也与同步复制还是异步复制无关。
 


 
Leader Election
 


 
引入Replication之后，同一个Partition可能会有多个副本（Replica），而这时需要在这些副本之间选出一个Leader，Producer和Consumer只与这个Leader副本交互，其它Replica作为Follower从Leader中复制数据。注意，只有Leader负责数据读写，Follower只向Leader顺序Fetch数据（N条通路），并不提供任何读写服务，系统更加简单且高效。
 


 
思考 为什么follower副本不提供读写，只做冷备？
 


 
follwer副本不提供写服务这个比较好理解，因为如果follower也提供写服务的话，那么就需要在所有的副本之间相互同步。n个副本就需要 nxn 条通路来同步数据，如果采用异步同步的话，数据的一致性和有序性是很难保证的；而采用同步方式进行数据同步的话，那么写入延迟其实是放大n倍的，反而适得其反。
 


 
那么为什么不让follower副本提供读服务，减少leader副本的读压力呢？这个除了因为同步延迟带来的数据不一致之外，不同于其他的存储服务（如ES，MySQL），Kafka的读取本质上是一个有序的消息消费，消费进度是依赖于一个叫做offset的偏移量，这个偏移量是要保存起来的。如果多个副本进行读负载均衡，那么这个偏移量就不好确定了。
 


 
TIPS
 
Kafka的leader副本类似于ES的primary shard，follower副本相对于ES的replica。ES也是一个index有多个shard（相对于Kafka一个topic有多个partition），shard又分为primary shard和replicition shard，其中primary shard用于提供读写服务（sharding方式跟MySQL非常类似：shard = hash(routing) % number_of_primary_shards。但是ES引入了协调节点(coordinating node) 的角色，实现对客户端透明。），而replication shard只提供读服务（这里跟Kafka一样，ES会等待relication shard返回成功才最终返回给client）。
 


 
有传统MySQL分库分表经验的同学一定会觉得这个过程是非常相似的，就是一个sharding + replication的数据架构，只是通过client(SDK)或者coordinator对你透明了而已。
 


 
Propagate消息
 


 
Producer在发布消息到某个Partition时，先通过ZooKeeper找到该Partition的Leader，然后无论该Topic的Replication Factor为多少（也即该Partition有多少个Replica），Producer只将该消息发送到该Partition的Leader。Leader会将该消息写入其本地Log。每个Follower都从Leader pull数据。这种方式上，Follower存储的数据顺序与Leader保持一致。Follower在收到该消息并写入其Log后，向Leader发送ACK。一旦Leader收到了 ISR (in-sync replicas) 中的所有Replica的ACK，该消息就被认为已经commit了，Leader将增加 HW( High-Watermark) 并且向Producer发送ACK。
 


 
为了提高性能，每个Follower在接收到数据后就立马向Leader发送ACK，而非等到数据写入Log中。因此，对于已经commit的消息，Kafka只能保证它被存于多个Replica的内存中，而不能保证它们被持久化到磁盘中，也就不能完全保证异常发生后该条消息一定能被Consumer消费。但考虑到这种场景非常少见，可以认为这种方式在性能和数据持久化上做了一个比较好的平衡。在将来的版本中，Kafka会考虑提供更高的持久性。
 


 
Consumer读消息也是从Leader读取，只有被commit过的消息（offset低于HW的消息）才会暴露给Consumer。
 


 
Kafka Replication的数据流如下图所示：
 

 


 
关于这方面的内容比较多而且复杂，这里就不展开了，这篇文章写的很好，有兴趣的同学可以学习
 
《 Kafka设计解析（二）：Kafka High Availability （上）》。
 


 
Kafka的几个游标(偏移量/offset)
 


 
下面这张图非常简单明了的显示kafka的所有游标
 
（https://rongxinblog.wordpress.com/2016/07/29/kafka-high-watermark/）：
 


 

 


 
下面简单的说明一下：

 


 
0、ISR
 
In-Sync Replicas list，顾名思义，就是跟leader “保存同步” 的Replicas。“保持同步”的含义有些复杂，在0.9版本，broker的参数replica.lag.time.max.ms用来指定ISR的定义，如果leader在这么长时间没收到follower的拉取请求，或者在这么长时间内，follower没有fetch到leader的log end offset，就会被leader从ISR中移除。ISR是个很重要的指标，controller选取partition的leader replica时会使用它，leader需要维护ISR列表，因此leader选取ISR后会把结果记到Zookeeper上。
 


 
在需要选举leader的场景下，leader和ISR是由controller决定的。在选出leader以后，ISR是leader决定。如果谁是leader和ISR只存在于ZK上，那么每个broker都需要在Zookeeper上监听它host的每个partition的leader和ISR的变化，这样效率比较低。如果不放在Zookeeper上，那么当controller fail以后，需要从所有broker上重新获得这些信息，考虑到这个过程中可能出现的问题，也不靠谱。所以leader和ISR的信息存在于Zookeeper上，但是在变更leader时，controller会先在Zookeeper上做出变更，然后再发送LeaderAndIsrRequest给相关的broker。这样可以在一个LeaderAndIsrRequest里包括这个broker上有变动的所有partition，即batch一批变更新信息给broker，更有效率。另外，在leader变更ISR时，会先在Zookeeper上做出变更，然后再修改本地内存中的ISR。
 


 
1、Last Commited Offset
 
Consumer最后提交的位置，这个位置会保存在一个特殊的topic：_consumer_offsets 中。
 


 
2、Current Position
 
Consumer当前读取的位置，但是还没有提交给broker。提交之后就变成Last Commit Offset。
 


 
3、High Watermark(HW)
 
这个offset是所有ISR的LEO的最小位置（minimum LEO across all the ISR of this partition），consumer不能读取超过HW的消息，因为这意味着读取到未完全同步（因此没有完全备份）的消息。换句话说就是：HW是所有ISR中的节点都已经复制完的消息.也是消费者所能获取到的消息的最大offset（注意，并不是所有replica都一定有这些消息，而只是ISR里的那些才肯定会有）。
 


 
随着follower的拉取进度的即时变化，HW是随时在变化的。follower总是向leader请求自己已有messages的下一个offset开始的数据，因此当follower发出了一个fetch request，要求offset为A以上的数据，leader就知道了这个follower的log end offset至少为Ａ。此时就可以统计下ISR里的所有replica的LEO是否已经大于了HW，如果是的话，就提高HW。同时，leader在fetch本地消息给follower时，也会在返回给follower的reponse里附带自己的HW。这样follower也就知道了leader处的HW(但是在实现中，follower获取的只是读leader本地log时的HW，并不能保证是最新的HW)。但是leader和follower的HW是不同步的，follower处记的HW可能会落后于leader。
 


 
Hight Watermark Checkpoint
 


 
由于HW是随时变化的，如果即时更新到Zookeeper，会带来效率的问题。而HW是如此重要，因此需要持久化，ReplicaManager就启动了单独的线程定期把所有的partition的HW的值记到文件中，即做highwatermark-checkpoint。
 


 
4、Log End Offset(LEO)
 
这个很好理解，就是当前的最新日志写入（或者同步）位置。
 


 
四、Kafka客户