Kafka消息阻塞

转自:http://jis117.iteye.com/blog/2279519

1. hi all：  
2.       大家都很关心kafka消息阻塞的情况（感谢RoctetMQ给我们的教训）。Kafka上线也有一段时间了，确实有出现过消息阻塞的情况，虽然不影响业务而且用临时办法解决了，但是我觉得可以跟大家总结一下。为了不引起大家的恐慌，我决定先把结论写出来：comsumer 非正常的rebalancing（重新分配分区）才会导致消费阻塞，如果不出现rebalancing，消息是不是重复消费或阻塞。  
3.   
4.       以下是这两个BUG的描述，这可能需要一些Kafka的知识，我会说得通俗一点，同也会留下一些参考给有兴趣的童鞋进一步了解。  
5.      1. 消费者处理过慢可能会导致重复消费  
6.      线上场景：BPM会订阅消费然后把流程信息一条一条索引到ElasticSearch，当索引处理较慢（30s）的时候会出现。  
7.      重现步骤：https://gist.github.com/richard2011/23b563e6ee5bad4e9d56 ，很普通的代码，消费代码段加上sleep(30s)。  
8.      产生原因：Kafka是使用poll()(长轮询)拉取消息，流程可以简单理解为: 拉取消息->向kafka broker发送心跳->提交 offset。当消费者处理过慢（session timeout为30s）没有向kafka broker发送心跳，而且没有提交 offset，broker就会发起rebalancing，这个分区就会分配给其他消费者重复消费。最坏的情况是一直在rebalancing，新的消息都不会被消费。  
9.      临时办法：排查发现线上业务线正常场景没有消费处理过慢的场景，processor组件维护了一个线程池，每条消息都用一个线程处理。tasker-center会不断地把消息放在java的ArrayBlockingQueue。其他周边后端服务（如存储）这种处理能力难以评估的，可以尝试一下先把消息缓存到本地队列再做批量插入的操作，其实很多日志类或大数据类使用Kafka都是这样做的，如Flume。  
10.      如果确实有这样的场景，请联系我，可以通过通过两个参数减小这问题发生，但都不是完美解决问题。1) 增加session time的时间，但同时也会增加客户端失败的时间。2）减小分区拉取值（max.partition.fetch.bytes默认为1M）, 但会影响吞量，而且以bytes为单位也无法评估消息的数量。  
11.      修复计划：Kafka社区专门针对这个问题写了篇WIKI https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=61333789 。这是他们的改进计划，内容有目的(Motivation), 计划修改(Proposed Change), 新增或修改的公共接口（New or Changed Public Interfaces），升级计划和兼容性（Migration Plan and Compatibility）和已放弃方案（Rejected Alternatives）。@听风，可以参考一下他的文档模板，用于轩辕组件的改进计划模板挺好的。这个BUG暂定为随着Kafka版本的升级来修复。  
12.   
13.       2. 多分区多consumer时rebalancing可能会导致某个分区阻塞。  
14.      线上场景：发生在在cart-processor，每个topic有18个分区，每个cart-processor有两个consumer(不同groupId)，8个cart-processor节点。当cart-processor发版（节点增加或删除）会引起rebalancing，这可能导致个topic的分区阻塞。  
15.      重现步骤：代码https://gist.github.com/richard2011/d92caaa4af50331b0953，创建18个分区的topic, 通过不断地增加或删除，同时通过kafka-consumer-groups.sh命令查看分区情况，直到出现分区阻塞。  
16.      产生原因：kafka client的BUG（ https://issues.apache.org/jira/browse/KAFKA-2978 ），Kafka github主分支已修复，但是还没有release版，越多分区和consumer越容易出现这个问题。  
17.      临时办法：每次使用kafka的程序发版时，用kafka-consumer-groups.sh命令查看分区情况，发现分区阻塞则重启对应的机器。同时也写了小工具，使comsumer再次rebalancing。  
18.      修复计划：观察线上出现的频率，如果频繁出现，将会修复kafka client代码，出现不频繁则随Kafka升级修复。  
19.