目录
1、生产者重复发送消息
生产者幂等性
生产者事务消息
2、消费者重复消费消息
导致重复消费的情况
1、生产者重复发送消息
2、消费者重复消费消息
1、生产者重复发送消息
生产者发送了消息，没有收到broker正确的应答响应，导致生产者重试。
生产者发出一条消息，broker落盘后因为网络原因，发送端得到一个发送失败的响应或超时网络中断，然后生产者收到一个可恢复的Exception，重试发送消息导致重复。
重试过程
1、new KafkaProducer() 后创建一个后台线程 KafkaThread 负责发送消息到Kafka集群
2、调用 KafkaProducer.send() 发送消息，实际上只是把消息保存到 RecordAccumulator 中；
3、RecordAccumulator 某一个批次被填满或者等待时间超过了所设定的阈值的时候，此时KafkaThread 线程就开始拉取消息，发送消息到Kafka集群
4、如果发送成功，那么返回成功；
5、如果发送失败，那么判断是否允许重试。如果不允许重试，那么返回失败的结果；如果允许重试，把消息再保存到 RecordAccumulator 中，等待后台线程 KafkaThread 再次发送。
在Kafka中消息存在三种语义：
1、最多一次（At most once）：消息可能会丢失，但绝不会被重复发送 ----> Producer禁止重试
2、至少一次（At least once）：消息不会丢失，但有可能被重复发送 ----> Producer开启了重试机制，当发送的消息没有收到Kafka集群成功的应答时会再次发送数据
3、仅有一次（Exactly once）：消息不会丢失，也不会被重复发送 ----> Kafka通过幂等性和事务这两个机制保证了精准一次
生产者幂等性
kafka为了实现幂等性，在 0.11.0 版本之后，它在底层设计架构中引入了ProducerID和SequenceNumber。
1、ProducerID：在每个新的 Producer 初始化时，会被分配一个唯一的 ProducerID，这个 ProducerID 对客户端使用者是不可见的。
2、SequenceNumber：对于每个 ProducerID，Producer发送数据的每个 Topic 和 Partition 都对应一个从 0 开始单调递增的 SequenceNumber 值。
异常情况如下图所示：
当 Producer 发送消息 (x2,y2) 给 Broker 时，Broker 接收到消息并将其追加到消息流中。此时，Broker 返回 Ack 信号给 Producer 时，发生异常导致
Producer 接收 Ack 信号失败。对于 Producer 来说，会触发重试机制，将消息 (x2,y2) 再次发送，
但是，由于引入了幂等性，在每条消息中附带 PID(ProducerID) 和 equenceNumber。相同的 PID 和 SequenceNumber 发送给 Broker，
而之前 Broker 缓存过之前发送的相同的消息，那么在消息流中的消息就只有一条 (x2,y2)，不会出现重复发送的情况。
注意：幂等生产者只能保证单个主题的单一分区内消息不重复；其次，它只能实现单会话上的幂等性，不能实现跨会话的幂等性，这里的会话即可以理解为：Producer 进程的一次运行。当重启了 Producer 进程之后，则幂等性保证就失效了。
如果我想实现多分区以及多会话上的消息无重复，应该怎么做呢？答案就是事务！
生产者事务消息
Kafka的事务特性本质上是支持了Kafka跨分区和Topic的原子写操作。通过事务机制，KAFKA 可以实现对多个 topic 的多个 partition 的原子性的写入，即处于同一个事务内的所有消息，不管最终需要落地到哪个 topic 的哪个 partition, 最终结果都是要么全部写成功，要么全部写失败（Atomic multi-partition writes）
Kafka事务的回滚，并不是删除已写入的数据，而是将写入数据的事务标记为 Rollback/Abort 从而在读数据时过滤该数据。
// @Transactional // 方式一需要添加该注解，但是添加该注解事务会自动回滚 @Test public void sendMsgTransaction() throws ExecutionException, InterruptedException { // 方式一 /*kafkaTemplate.send("topic-02" , 0 , null , "msg11....") ; int a = 1 / 0 ; kafkaTemplate.send("topic-02" , 1 , null , "msg12....") ;*/ // 方式二 kafkaTemplate.executeInTransaction(operations -> { operations.send("topic-02" , 0 , null , "msg09....") ; int a = 1 / 0 ; operations.send("topic-02" , 1 , null , "msg10....") ; return true ; }) ; }2、消费者重复消费消息
# 解决方案
1、消费端手动移位提交
2、消费端做幂等处理
# 幂等操作
1、新增时，先查一下，如果有了就不操作了
2、更新，需要set为某个值，不需要处理，天然幂等
3、生产者每次传递一个唯一id，消费消息时，先去redis中查一下是否已经消费，如果没有，就消费消息，然后把id存入到redis中；如果存在，就不用处理了
4、基于数据库的唯一索引， 新增的数据在数据库中存在重复的直接报错，不会出现脏数据
