Kafka Consumer(消费者组)

Posted 2021-02-18 snail-gao

消费者组（consumer group）

官方定义：
消费者使用一个消费者组（即group.id）来标记自己，topic的每条消息都只会被发送到每个订阅它的消费者组的一个消费者实例上。

基本概念

- 所有consumer实例都属于相同group--实现基于队列的模型。每条消息只会被一个consumer实例处理。
- consumer都属于不同group--实现基于发布/订阅的模型。极端的情况是每个consumer实例都设置完全不同的group，这样Kakfa消息就会被广播到所有consumer实例上。
- kafka目前只提供单个分区的消息顺序，而不会维护全局的消息顺序，如果要实现topic全局的消息顺序，只能通过让每个consumer group下只包含一个consumer实例的方式来间接实现。

总结

- consumer group下可以有一个或者多个comsumer实例。一个consumer实例可以是一个线程，也可以是其他机器上的进程。
- group.id是consumer group唯一标识。
- 对于某个group而言，订阅topic的每个分区只能分配给该group下的一个consumer实例（当然该分区还可以被分配给其他订阅该topiv的消费组）

位移

位移：kafka在内部采用一个map来保存其订阅topic所属分区的offset。
位移提交：旧版本的位移提交时提交到Zookeeper上面的固定节点上。该路径是/consumer/<group.id>/offsets/<topic>/partitonId。
新版本的位移提交到kafka内部的一个topic（__consumer_offsets）上。

消费者组重平衡

只对消费者组(consumer group)有效，如果是独立消费者(standlone consumer)，没有rebalance的概念。
rebalance：consumer group下所有consumer如何达成一致来分配订阅topic的所有分区。假设有一个consumer group有20个consumer实例。该group订阅了一个具
有100分区的topic。正常情况下每个consumer会分配5各分区。

consumer主要参数

session.timeout.ms: 检测consumer group组成员发送崩溃的时间("coordinator检测失败的时间")。如果有消息需要很长时间，那么consumer有可能无法执行任何消费。
在0.10.1.0版本kafka对该参数做了拆分。可以指定一个比较小的值让coordinator能够快速检测consumer崩溃，开启rebalance。默认参数值10秒。
max.poll.interval.ms: 设置消息处理逻辑的最大时间。
auto.offset.reset: 指定了无位移信息或位移越界是kafka的应对策略。
- earliest: 指定从最早的位移开始消费，这里最早的位移不一定是0.
- latest:指定从最新处位移开始消费。
- none:指定如果未发现位移信息或位移越界，则抛出异常。很少使用。
enable.auto.commit: 是否自动提交。
fetch.max.bytes: 如果实际业务消息很大，必须要设置该参数为一个较大的值，不然会无法消费这些信息。
max.poll.records: 控制单次poll调用返回的最大消息数。默认500.
heartbeat.interval.ms: 当开启新的一轮rebalance是，他会将这个以REBALANCE_IN_PROGRESS异常的形式“塞进”consumer心跳请求中，这样其他成员拿到response后
才知道它需要重新加入group。这个参数设置时间必须小于session.timeout.ms。
connections.max.idle.ms: kakfa会定期的关闭空闲Socket连接，默认9分钟，设置-1时，即不管吧这些空闲连接。

订阅topic

consumer订阅是延迟生效的，只有在下次poll请求时才会生效。