Kafka消费和Offset提交流程

更新偏移量

是

json

ListenerConsumer

ConsumerAcknowledgment

非自动提交 + 非Record AckMode

批量拉取数据pollTimeoutpollCount

  查询账户余额

acks：待Ack的Offset的阻塞队列

默认：AckMode=Batch，ConsumerMode=非Batch一、AckMode：Record（每条记录）、Batch（批量）、Time（时间）、COUNT（待Ack数量）、COUNT_TIME（次数或时间）、MANUAL（手动）、MANUAL_IMMEDIATE（手动立即提交）二、Spring kafka ConsumerModeBatch（批量）、非Batch（单个）

修改数据库

非手动立即提交

批量执行

和当前偏移量比，大于当前偏移量才会更新

是否为当前线程

Note：前提 isAutoCommit为false1）由Spring管理我们的自动提交2）手动提交（手动提交和手动立即提交）

AckMode = Record？

否

提交偏移量

系统

循环执行

ConsumerBatchAcknowledgment

userid

业务手动提交

1、KafkaListenerAnnotationBeanPostProcessor.postProcessAfterInitialization()2、KafkaListenerEndpointRegistrar.registerEndpoint()     KafkaListenerEndpointRegistrar.createListenerContainer()3、AbstractKafkaListenerContainerFactory.createListenerContainer()    ConcurrentKafkaListenerContainerFactory.createContainerInstance()5、ConcurrentMessageListenerContainer    AbstractMessageListenerContainer实现了 SmartLifeCycle 生命周期方法    在Bean初始化后，其onStartUp方法如果返回True，就会调用其start()方法    ConcurrentMessageListenerContainer.doStart()

用户开户

业务onMessage(List<Record>)

非自动提交 + 非任何Manual手动提交模式

业务onMessage(Record)

获取偏移量

是否MANUAL_IMMEDIATE

2、更新

获取records中最大偏移量记录

拉取阻塞队列数据

循环拉取

更新

1、拉取

Spring Batch模式提交（处理上个循环的记录待提交的Offset）

调用银行

TIME和TIME_COUNT模式提交偏移量条件是否满足

加入阻塞队列

3、提交逻辑

Spring Batct模式提交

是否批量消费

非手动提交

用户

Kafka

AckMode非TIme 切非Cout、