Kafka Consumer消费以组的方式划分,Topic中的每一个分区只会分给同一个组中的其中一个实例。这是基于队列模式,如果想基于发布订阅模式,那订阅同一个Topic的实例需要指定不同的组名。
必需参数bootstrap.serversKafka服务器
group.idConsumer Group的名字,唯一标识一个consumer group
key.deserializerKey的反序列化,二进制的消息Key转换成具体的类型
value.desrializerValue的反序列化,二进制的消息内容转换成具体的类型
主要参数session.timeout.mscoordinator检测失败的时间,通常需要设置一个较小的值,这样可以快速检测到consumer崩溃的情况,尽快开启rebalance。
max.poll.interval.ms用于设置消息处理逻辑的最大时间
auto.offset.resetconsumer group无位移信息和位移越界时Kafka对应的策略。consumer group重启不会使用该策略,因为Kafka已经记录了group的唯一信息
earliest:从最早的位移开始消费,不一定就是0latest:从最新位移处开始消费none:如果无位移信息和位移越界,抛出异常。enable.auto.commit指定consumer是否自动提交位移,默认为true
fetch.max.bytes指定consumer单次获取数据的最大字节数
max.poll.records控制poll方法返回的最大消息数量
heartbeat.interval.ms控制consumer group中成员感知rebalance的时间。
connections.max.idle.ms空闲连接空闲时间超过该参数,会被关闭。
auto.commit.interval.ms后台自动提交位移的时间间隔
消息轮询Poll新版Consumer采用了类似Linux I/O模型Poll,使用一个线程管理多个socket连接,然后循环Poll消息。
poll方法返回的条件是要不获得了足够多的数据,或者超过了指定的超时时间。
位移管理新版本的consumer位移已交由内部topic管理(_consumeroffsets),该Topic有多个分区,每个分区有多个副本(可以通过参数控制)。该内部Topic存在的唯一目的保存consumer提交的位移。
手动提交位移支持同步和异步,提交需要位移需要指定一个Map,key是TopicPartition,value是OffsetAndMetadata,里面存储了下一条待消费消息的offset。
---来自腾讯云社区的---shysh95
微信扫一扫打赏
支付宝扫一扫打赏