1.前言

本次的案例和相关配置是基于spring ，具体的版本如下

spring.5.1.5
Spring-kafka-2.2.4
spring-clients-2.0.1

2.消息丢失

2.1 消息为什么会丢失

2.1.1 消费相关参数说明

enable.auto.commit：表示消费者会周期性自动提交消费的offset。默认值true。
auto.commit.interval.ms：在enable.auto.commit为true的情况下， 自动提交的间隔。默认值5秒。
max.poll.records：单次消费者拉取的最大数据条数，默认值500。
max.poll.interval.ms：表示若在阈值时间之内消费者没有消费完上一次poll的消息，
consumer client会主动向coordinator发起LeaveGroup请求，触发Rebalance（再平衡调整）；
然后consumer重新发送JoinGroup请求。
session.timeout.ms：group Coordinator（协调者）检测consumer发生崩溃所需的时间。
在这个时间内如果Coordinator未收到Consumer的任何消息，那Coordinator就认为Consumer挂了。默认值10秒。
heartbeat.interval.ms：标识Consumer给Coordinator发一个心跳包的时间间隔。heartbeat.interval.ms越小，发的心跳包越多。默认值3秒

2.1.2 案例

enable.auto.commit：true 表示消费者会周期性自动提交消费的offset。

如果消费者使用以上的配置，就可能会出现消息丢失，大概流程如下；

2.2 解决办法

手动提交offset.具体的代码如下

2.2.1 消费者配置文件

参数解释：

earliest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
latest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
none
topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

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,
}

2.2.2 监听器示例代码

3. 消息重复消费

3.1 消息重复消费出现的原因

原因1：

消费者宕机、重启或者被强行kill进程，导致消费者消费的offset没有提交。

原因2：

设置enable.auto.commit为true，如果在关闭消费者进程之前，取消了消费者的订阅，则有可能部分offset没提交，下次重启会重复消费。

原因3：

消费后的数据，当offset还没有提交时，Partition就断开连接。比如，通常会遇到消费的数据，处理很耗时，导致超过了Kafka的session timeout.ms时间，那么就会触发reblance重平衡，此时可能存在消费者offset没提交，会导致重平衡后重复消费。

3.2 消息重复消费解决办法

方法1：

思路是提高消费能力，提高单条消息的处理速度，例如对消息处理中比 较耗时的步骤可通过异步的方式进行处理、利用多线程处理等。在缩短单条消息消费时常的同时，根据实际场景可将max.poll.interval.ms值设置大一点，避免不 必要的rebalance，此外可适当减小max.poll.records的值，默认值是500，可根 据实际消息速率适当调小。这种思路可解决因消费时间过长导致的重复消费问题， 对代码改动较小，但无法绝对避免重复消费问题。

方法2：

思路是引入单独去重机制，例如生成消息时，在消息中加入唯一标识符如消息id等。在消费端，我们可以保存最近的1000条消息id到redis或mysql表中，配置max.poll.records的值小于1000。在消费消息时先通过前置表去重后再进行消息的处理。