kafka手动提交offset以及offset回滚

kafka手动提交offset以及offset回滚 1 手动提交

当 auto.commit.enable 设置为false时，表示kafak的offset由消费者手动维护，spring-kafka提供了通过ackMode的值表示不同的手动提交方式；

ackMode有以下7种值：

public enum AckMode

        // 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
        RECORD,

        // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
        BATCH,

        // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
        TIME,

        // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
        COUNT,

        // TIME | COUNT 有一个条件满足时提交
        COUNT_TIME,

        // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
        MANUAL,

        // 手动调用Acknowledgment.acknowledge()后立即提交
        MANUAL_IMMEDIATE,
    }

如果设置 AckMode 模式为 MANUAL 或者 MANUAL_IMMEDIATE，则需要对监听消息的方法中，引入 Acknowledgment 对象参数，并调用 acknowledge() 方法进行手动提交；如果没有提交程序重启后会重复消费；其他模式满足条件就会提交（不是严格意义上的手动），不需要调用acknowledge() 。

2 回滚offet

处理数据报错时记下offset，重置offset就会重复消费。

 TopicPartition topicPartition = new TopicPartition(topic, partition);
 consumer.seek(topicPartition, offSet);