消息的重复消费在生产端和消费端都可能发生，下面一一讲解：

①：生产端消息重复发送
        发送消息如果配置了重试机制，比如由于网络波动，生产者未得到broker收到了消息的响应，就会触发重试机制，3秒后再次发送此消息。broker之前已经收到过这个消息，但生产者由于触发了重试机制，就导致了消息的重复发送。那么broker就会在磁盘缓存多条同样的消息，消费端从broker拉取消息时，就会造成重复消费。

注意：kafka新版本已经在broker中保证了接收消息的幂等性（比如2.4版本），只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可，默认是false不开启。

新版kafka的broker幂等性具体实现原理：
        kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker，broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和Sequence Number，如果相同不会再接收。

1. PID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个PID
   对用户完全是透明的。生产者如果重启则会生成新的PID。
2. Sequence Number：对于每个 PID，该 Producer 发送到每个 Partition 的数据都有对应的序列号，这些序列号是从0开始单调递增的。

①：消费端消息重复消费
        对于消费端消息的重复消费问题，如果消费端拉取了一部分数据，消费完毕后，准备执行手动提交（或自动提交）时，消费者挂掉了！此时offset还未提交呢，那么当服务重启时，还是会拉取相同的一批数据重复处理！造成消息重复消费

无论是生产者还是消费者的重复消息，一般都会在消费端卡死，做幂等性处理。
        幂等性可以用redis的setnx分布式锁来实现。比如操作订单消息，可以把订单id作为key，在消费消息时，通过setnx命令设置一下，offset提交完成后，在redis中删除订单id的key。setnx命令保证同样的订单消息，只有一个能被消费，可有效保证消费的幂等性！

        kafka想要保证消息顺序，是需要牺牲一定性能的，方法就是一个消费者，消费一个分区，可以保证消费的顺序性。但也仅限于消费端消费消息的有序性，无法保证生产者发送消息有序。

        比如：如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现，发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1了。发送端消息发送已经乱序，到了消费端消费时，自然无法保证顺序！

        如果一定要保证生产-消费全链路消息有序，发送端需要同步发送，ack回调不能设置为0。且只能有一个分区，一个消费者进行消费，但这样明显有悖于kafka的高性能理论！

问题：如何在多个分区中保证消息顺序和消息处理效率呢？

        首先使用多个分区，消息可以被发送端发送至多个分区，保证消息发送的效率。然后在消费端在拉消息时使用ConutdownLunch来记录一组有序消息的个数。如果达到个数，说明已拉取到完整的一组有序消息。然后在消费端根据消息序号进行排序，消费端将排好序的消息发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。即可最大程度上既保证顺序又保证效率！

线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。
解决方案：此种情况如果积压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到的消息快速转发到其他topic(可以设置很多分区)，然后再启动多个消费者同时消费新主题的不同分区。如图所示：

由于消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。
解决方案：此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)，后面再慢慢分析死信队列里的消息处理问题。这个死信队列，kafka并没有提供，需要整合第三方插件！

延时队列存储的对象是延时消息。所谓的“延时消息”是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多， 比如 ：

1. 在订单系统中， 一个用户下单之后通常有 30 分钟的时间进行支付，如果 30
   分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了。
2. 订单完成1小时后通知用户进行评价。

但kafka没有提供延时消息功能，可以用rocketmq、rabbitmq都做延时消息。如果一定要用kafka实现延时消息呢？

实现思路：发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中（topic_1s，topic_5s，topic_10s，…topic_2h，这个一般不能支持任意时间段的延时），然后通过定时器进行轮训消费这些topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

        如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序bug导致的计算错误，当程序bug修复后，这时可能需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消费，这种可以用consumer的offsetsForTimes、seek等方法指定从某个offset偏移的消息开始消费，完成消息的回溯消费！