Offset的作用Kafka将消息存储在topic中每个topic可以包含多个partition。每个消费者从topic的特定partition读取消息。在此过程中Kafka借助offset来追踪消费者已读取的消息位置。offset是从零开始的整数序列每存储一条消息对应partition的offset就会递增1。假设某个消费者已从partition读取了5条消息消费完成以后需要把offset提交给Kafka服务端Kafka的服务端就会将offset从0-4标记为已提交。当该消费者下次尝试读取消息时将从offset5开始继续消费。消费者的offset的提交总体来说分2种情况一种是每隔一段时间自动提交一种是按照AckMode进行提交。自动提交默认自动提交是关闭的我们可以通过配置进行打开spring:kafka:consumer:enable-auto-commit:trueauto-commit-interval:10senable-auto-commit是否启用自动提交默认是禁用auto-commit-interval自动提交的时间间隔比如这是我们的consumerKafkaListener(topicsquickstart-events)publicvoidonMessage(Stringmessage)throwsException{log.info(on message:{},message);}发送完消息以后立即停掉应用然后再重启应用消息就会再次投递过来因为还不到10soffset还没提交。按照AckMode提交目前支持7种不同的AckMode可以分为2大类一类是由Spring容器自动提交Spring会去监听业务逻辑的执行业务逻辑执行完以后如果不抛出异常Spring会帮着自动提交offset如果出了异常会走异常处理逻辑另一类是由Listener手动方式提交。自动方式RECORD当listener处理完当前记录后立即提交该记录的offset。BATCH当 poll() 方法返回的所有记录都处理完成后提交这批记录的offset这是默认的AckMode。TIME当 poll() 方法返回的所有记录都处理完成后且距离上次提交时间超过设定的ackTime时提交offset。COUNT当 poll() 方法返回的所有记录都处理完成后且自上次提交以来接收的消息数量达到ackCount阈值时提交offset。COUNT_TIME结合TIME与COUNT的混合模式只要满足任一条件时间间隔或消息数量即执行提交。手动方式MANUAL由listener负责调用Acknowledgment.acknowledge()方法。该方法调用后容器将按照BATCH模式的语义提交offset。MANUAL_IMMEDIATE当listener调用Acknowledgment.acknowledge()方法时立即提交当前offset。如果使用了事务offset会被发送到事务中其语义等同于 RECORD 或 BATCH 模式具体取决于listener类型单条记录监听器或批量监听器。还是上面的案例我们禁用掉自动提交修改我们的listener:KafkaListener(topicsquickstart-events)publicvoidonMessage(Stringmessage)throwsException{log.info(on message:{},message);Thread.sleep(10000);}发一条消息以后立即关闭应用重启以后消息会再次投递过来因为此时的AckMode是默认的BATCH业务还没处理完listener就挂掉了还没有机会去提交offset。但是如果等10秒以后再重启应用消息就不会再次投递过来因为offset已经提交了。自定义AckMode比如我们想改成AckMode.MANUAL_IMMEDIATEBeanpublicContainerCustomizerObject,Object,ConcurrentMessageListenerContainerObject,ObjectkafkaContainerCustomizer(){returncontainer-{container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);};}此时需要修改我们的listener手动去提交offsetKafkaListener(topicsquickstart-events)publicvoidonMessage(Stringmessage,Acknowledgmentack)throwsException{log.info(on message:{},message);Thread.sleep(10000);ack.acknowledge();}如果不想提交offset还可以KafkaListener(topicsquickstart-events)publicvoidonMessage(Stringmessage,Acknowledgmentack)throwsException{log.info(on message:{},message);ack.nack(Duration.ofSeconds(5));}此时每隔5秒消息就会重新投递过来一次。更多文章请关注公众号“爪哇优太儿”