消费者组与消费者偏移量

消费者组

具有相同组ID的消费者就属于同一个消费者组，它有如下特点：

• 一个组可以有多个消费者

• 主题中的消息只能被同一个组中的一个消费者消费

• 一个主题可以被多个消费者组消费

消费者组的概念主要是为了实现点对点队列模式和发布订阅模式，它是如何实现的呢？

• 队列模式：所有消费者属于同一个组

• 发布订阅模式：消费者属于不同组，也就是有不同消费者组

另外消费者组模式具有高伸缩性，通常有几个分区那么消费者组中就包含几个消费者，这样形成一对一模式，每个消费者都顺序的消费对应的分区（注意这里是分区消息顺序性并不是全局顺序性），如果某个消费者崩溃了，那么它消费的分区就会被分配给其他消费者，这样保证继续工作，这个过程叫做rebalance，这也是消费者组的一个好处，不过凡事有利有弊，rebalance频繁发生也是灾难。

消费者偏移量

__consumer_offsets主题

这里说的消费者偏移量指的是消费者消费某个主题的消费进度，消费者自己包括Kafka集群都需要知道这个进度，因为无论消费者是启动中还是重启后都需要知道自己应该从哪里继续消费。如何上报这个信息呢？其实就是位移提交，默认消费客户端是自动提交的，通常我们都会设置为手动提交。

切记：这个消费者offset指的是消费者要消费的下一条消息的位移，而不是当前消费到哪里了。

消费者偏移量其实并不复杂，无非就是记录消费者针对某个主题的消费进度，所以这个东西就是键值：

• 键：Group id + 主题 + 分区号

• 值：offset值

老版本的消费者偏移量都是放在Zookeeper上，新版本的消费者偏移量是放在单独的一个主题上，叫做__consumer_offsets，默认有50个分区，每个分区默认1个副本，如下图：

我这里是3台的kafka集群，所以这50个分区被分布在3台上。

offsets.topic.replication.factor=3 设置这个broker参数，默认为1，那么创建该主题的时候就会是3个副本。

在第一次搭建集群的时候不会创建这个主题，只有在第一次启动消费者程序之后kafka才会自动创建这个主题。

这个主题除了放消费者消费偏移量之外还会存放其他类型消息，保存消费者组的注册消息和删除Group过期位移消息，而删除其实就是根据键来保留最近的消息。

# 新版本查看消费者偏移量
kafka-consumer-groups.sh --bootstrap-server 172.16.100.10:9092 --describe --group TestGroup

# 老版本查看消费者偏移量
kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect [ZOOKEEPER_IPADDRESS]:[ZOOKEEPER_PORT] --group [CONSUMER_GROUP]

说明：50个__consumer_offsets分区，消费者到底要使用哪一个分区呢？这个就是就是找Coordinator组件，在后面会有这个组件的说明。

自动提交和手动提交的选择

自动提交：消费者第一次poll消息的时候会根据auto.offset.reset策略来决定从哪里开始消费，因为消费是个循环，就会不停的调用poll，至于什么时候提交呢？这是由auto.commit.interval.ms来决定的，它会定期去提交。如果出现Rebalance的情况，那么在Rebalance之后消费者则从上一次提交位移的地方继续消费。所以基于上面的特点，自动提交不会丢失消息（也就是消息都会被消费）但是会发生重复消费的情况。比如你已经消费了消息并执行了业务逻辑，但是还没有到达提交周期，这时候线程挂了，重启后还是从上次的位置消费这就出现了重复消费，当然也有可能出现消息丢失的情况。

手动提交：灵活度高不过这里有2种也就是同步提交和异步提交。两种各有利弊，同步提交会阻塞但是可以实现重试机制；异步提交虽然不会阻塞但是无法重试，因为重试时的位移可能已经不是之前的位移了。

这里要说一下语义保证：

• 最多一次：消息可能会丢失，但是不会出现重复消费的情况

• 最少一次：消息不会丢失，但是可能会出现重复消费情况

• 精确一次：消息一定会被处理且只会被处理一次

如果业务对于消息的丢失或者重复消费没有要求那么可以使用自动提交；如果要求精确一次的语义那么就要使用手动提交。

独立消费者

消费者组消费通常我们叫做subscribe而独立消费者也就是不使用组的都是需要手动分配要消费的分区以及手动管理偏移量，这种方式我们称作assign，但是这两种方式有区别：

subscribe()和assign()的区别：如果使用前者则必须使用消费者组因为在一个组内消费者消费的分区是自动分配的，这个是Kafka内部通过算法来完成所以有时候组内成员变化或者分区数量变化会出现reblance的情况，但总之会均衡分配给组内的消费者。而后者是手动指定主题的分区，所以这种消费方式无需消费者组，就算指定了也不起作用。至于偏移量的话使用消费者的形式偏移量无论是手动提交还是自动提交都会保存在broker上，下次可以继续消费；但是assign的方式都是客户端自己维护偏移量，而且我们是无法查看它的偏移量的，无论是命令行还是其他监控形式。除了特殊场景外一般都是使用消费者的形式。我在工作中的确遇到过需要手动指定分区消费的场景。

真实案例：公司一个程序它会有多个实例，每个实例使用消费组方式来消费同一个主题，但是每个程序的消费者组除了前缀一样，后面都会随机生成一个字符串，这就变成了其实每个程序都是属于一个单独的组，而且程序重启后组也就变了。这里面有几个逻辑。消费者偏移量提交毫无意义，因为组变了，所以使用消费者组也毫无意义。大量没用组的脏数据留在集群上。但他们的需求是每个程序都全量消费所有分区，而且配置的是从最新点位消费。不过这里存在一个问题就是，如果4个程序，同时宕机1分钟，然后启动，就会存在丢失消息的情况，因为中途1分钟的消息它们是不会拿到的。因为组名变量，位移丢失了。

解决方案是：不使用消费组，改用assign的方式，首先获取指定主题的分区，然后通过assign分配分区，seek_to_end，剩下的就和普通消费一样了。不需要提交位移，在程序运行期间客户端会自己在内存中维护位移以实现持续消费，但是重启后就不行了。这样就满足了上面业务的场景要求。不过需要注意的是需要关注分区变化，如果变化
了就需要重新assign一下。另外assign方式消费是无法监控的。只能客户端自己打日志了。assign的方式一般用在流计算或者某些类广播场景。正常情况下，不建议使用assign的。

消费者组重平衡（rebalance）

rebalance的本质就是让某一个消费者组中的成员可以均衡的消费主题，所以这是一个分配过程，引发这个rebalance的条件通常有3个：

• 消费者组成员变化（新加入成员，已有消费者崩溃，消费者在一定时间内没有完成消息处理也会被视为崩溃）

• 消费者组所消费的主题分区增加（分区不允许减少）

• 组订阅的主题数量发生变化

虽然这个机制是为了保证组内消费者均衡的消费主题而不出现忙闲不均的情况，不过这个机制最遭人诟病的就是STW，JVM垃圾回收机制导致，在垃圾回收期间所有应用线程都会停止工作，所以你想想。而且如果你的消费者组有很多消费者，这个rebalance的过程还会消耗大量时间，而且它还会影响消费者的TPS。总之经常发生rebalance不是一件好事情，所以我们要尽量避免，注意是尽量避免因为你无法做到不发生而且。

我们这里说的避免是说非人为干预的，如果我们人为的增加了消费者组成员那一定会进行rebalance。

在消费者程序中我们可以设置如下参数来尽量避免发生rebalance，下面以Python客户端为例，对于Java客户端也是一样的：

# 默认10000毫秒，回话超时时长，Coordinator在10秒内没有收到Group下某个
# Consumer的心跳，则该Consumer被认为挂了，那么broker则开始rebalance
session_timeout_ms

# 消费者向Coordinator发送心跳的频率，默认3秒。
heartbeat_interval_ms

# 消费者两次poll的最大时间间隔，默认30秒。表示在这个时间内消费者无法完成poll方法的返回，那么消费者则被认为挂了，也会进行rebalance
max_poll_interval_ms

session_timeout_ms和heartbeat_interval_ms并不冲突，通常后者要小于前者，后者是周期，前者是N个周期。一般我们把heartbeat_interval_ms设置为2000，把session_timeout_ms设置为6000，也就是 Coordinator在6秒内（3个心跳周期）收不到心跳则认为消费者挂了。所以这两个参数是避免因为网络抖动而心跳信息没有及时发出导致超时而引发的rebalance。不过session_timeout_ms也不能设置太长，因为这样就无法及时发现真正挂了的消费者。

而max_poll_interval_ms解决的是因为消息处理时间过长其实消费者并没有挂但是达到了max_poll_interval_ms设置的值从而被误认为是消费者挂了而引发的rebalance。所以这个参数到底设置多少你要去评估一下每次获取消息的条数以及处理这些消息大约话费多长时间。

关于Coordinator

这是一个协调组件，它负责执行消费者注册、组成员管理包括执行rebalance和元数据操作。消费者提交位移也是向Coordinator所在的broker提交。每一个broker都有Coordinator，那么消费者如何找到属于自己的Coordinator呢？这里就是靠__consumer_offsets主题来完成，主要分2步：

• kafka会计算group.id的哈希值

• 计算__consumer_offsets的分区数量，默认是50个

• abs(哈希值 % 50) 也就是取模求绝对值，这个值就是该消费者要使用的__consumer_offsets主题的分区编号

当知道了__consumer_offsets的分区编号，那么该分区的Leader副本所在的broker就是该消费者要使用的Coordinator。

如何监控消费者消费进度

通过命令查看

kafka-consumer-groups.sh --bootstrap-server 192.168.5.138:9092 --describe --group TestGroup

另外就是通过JMX监控，这种形式可以集成到监控系统中。这里有2个指标比较重要：

• records-lag-max：这个表示积压量，值越大表示消费者越落后生产者。

• records-lead-min：表示消费者最新消息位移与分区当前第一条消息位移的差值。

这个被监控的JMX是消费者程序而不是Kafka服务器的，而且只能是JAVA消费者程序。

标签：组与,消费,消费者,分区,偏移量,提交,rebalance
来源： https://www.cnblogs.com/yinging/p/16512815.html