【Kafka】从Consumer消费能力低下谈起

近期在生产环境发下日志入库延迟，导致很多准实时的监控图表获取不到信息，这问题以前没有出现过，可能跟最近业务量上升有关，毕竟日均小两亿的平台了。梳理系统架构发现，日志是缓存在kafka中，由一个后台进程task从kafka中消费，存放到数据库中的，日志入库延迟，跟task关系很大。由于之前对kafka了解很少，以此为契机进行下学习，并以此博客为记录，温故知新。

什么是Kafka呢？
官方说kafka是一个高吞吐量的分布式发布订阅消息系统。它有很多术语，比如producer，consumer，broker，如下简单解释下。

Producer：消息的生产者，向kafka发送消息的客户端

Consumer：消息的消费者，从kafka接取消息的客户端

Broker：kafka集群中的单个实例

Group：特指consumer group，多个consumer可以组成一个组，每个消息只能被组中的一个consumer消费，如果一个消息可以被多个consumer消费的话，那么这些consumer必须在不同的组，这个特性很重要

Topic：kafka主题，一个Topic可以认为是一类消息

Partition：分区，一个topic又可以分成多个分区，一个分区的消息只能被组内一个consumer消费

Offset：偏移量，offset为一个long型数字，它是唯一标记一条消息。它唯一的标记一条消息，我把他理解成游标卡尺的游标，标记消息的位置

接着我们来看一个简单的kafka系统结构：

需要补充的是，kafka本身不维护producer和consumer的信息，这些信息是由zookeeper来保存维护的，所以producer和consumer的加入离开，不会影响到kafka集群本身

kafka一些特点

消息状态：

消息状态保存在consumer中，broker不关心消息是否消费被谁消费了，只会记录一个偏移量（offset），正常情况下，broker中消息将会被顺序消费，consumer驱动offset线性向前移动

支持批量发送：

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

push-and-pull :

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

分区机制partition：

Kafka的broker端支持消息分区，Producer可以决定把消息发到哪个分区，在一个分区中消息的顺序就是Producer发送消息的顺序，一个Topic中可以有多个分区，具体分区的数量是可配置的。

Producer将消息发布到指定的Topic中,同时Producer也能决定将此消息归属于哪个partition;比如基于”round-robin\"方式或者通过其他的一些算法等。本质上kafka只支持Topic.每个consumer属于一个consumer group;反过来说,每个group中可以有多个consumer.发送到Topic的消息,只会被订阅此Topic的每个group中的一个consumer消费。这一句很重要，只会被订阅topic的每个group中的一个consumer消费！

上面提到，一个topic的消息，会在一个group中的consumer中负载均衡。一个Topic中的每个partions,只会被一个\"订阅者\"中的一个consumer消费,不过一个consumer可以消费多个partitions中的消息。

partitions和consumer之间的关系如下：

设定一个topic
partition情形: p0, p1, p2, p3，
consumer情形（同group）: c0, c1,c2
那么
如果按范围分配策略，分配结果是： c0: p0, c1: p1, c2: p2, p3
如果按轮询分配策略，分配结果是： c0: p1,p3, c1: p1, c2: p2

分配过程：

常见问题
回到本文最开始的引子，当时后台进程日志中出现如下报错提示：
Attempt to heart beat failed since the group is rebalancing, try to re-join group.
Error UNKNOWN_MEMBER_ID occurred while committing offsets for group 1
Auto offset commit failed: Commit cannot be completed due to group rebalance
Error UNKNOWN_MEMBER_ID occurred while committing offsets for group 1
Auto offset commit failed:
Attempt to join group 1 failed due to unknown member id, resetting and retrying.
提示很明显了，心跳检测失败，kafka认为这个consumer已经挂掉了，于是将其移除group，重新对partitions和组员进行分配（rebanlence），但实际上这个consumer还在处理，待它处理完提交offset时，发现，哎，我的分区呢？失败！

对于这个问题，主要思路有如下几个：
1，提高心跳检测的超时时间，session.timeout.ms:超时时间设置大一点
2，减小每次取kafka的消息大小，max.partition.fetch.bytes:一次拉取最大byte这个属性 小一点.默认1M
3，提高partitions的实例数量（和consumer匹配，以>=consumer数量为宜）
4，提高consumer效率，取数据和处理数据异步，consumer中不要有额外的时间消耗，以免影响取数据或心跳检测
5，在网上资料中，还有提到，如果用的是spring-kafka的，可以关闭掉自动提交的设置enable.auto.commit