Kafka、RocketMQ、RabbitMQ消息丢失可能存在的地方，以及解决方案

这里主要对比：Kafka、RocketMQ、RabbitMQ

介绍一下消息生产、存储、消费三者的架构形式。

消息丢失可能存在的场景：

情况 1： 生产者发送给MQ的过程消息丢失

在写消息的过程中因为网络的原因，还没到mq消息就丢失了；或者是到了mq，还没返回确认收到消息，消息丢失了，怎么处理？

情况 2：【补充】生产者在准备批量推送给MQ时，先缓冲到jvm内存，如先放到list集合中，凑够一定数量再批量推送，此时如果这台服务器宕机，就会面临丢失情况，怎么处理？

情况 3 ： Broker存储消息丢失

mq的master节点在接受到消息后，还没来得及同步给slave上面的follower副本，master节点直接宕机了，消息还在内存中，丢失消息了，怎么处理？

情况 4 ： 消费者拉取消息后，消息丢

消费者拉取到了消息，但是还没来得及消费，消费者实例挂掉了，mq以为这条消息已经消费了，怎么处理？

针对“情况 1”的解决思路：

方法1：

AMQP很多都支持事务机制，那是不是说我们可以生产者采用同步模式给mq发送消息的过程，打开事务机制，发送一条，mq接受到后返回确认信息后再发送下一条消息？这样肯定行不通的，会使生产者的吞吐量直线下降。

方法2：

对于 RabbitMQ：

它提供了一个异步confirm机制，发完了生产者就不用管了，rabbitmq接收到消息会回调生产者的本地一个接口告诉你消息接收成功了，如果rabbitmq在接收消息是报错了，就会回调这个接口，告诉你这个消息接收失败了，可以重试发送了，不会阻塞下一条消息的发送。

对于RocketMQ：它提供了一个异步接收模式， Async

API，Producer只需要把消息发送给mq就可以。mq处理完成之后，调用producer定义好的回调函数，就可以确认是否收到对应messageID的回调确认消息，没有收到就触发retry机制。

对于Kafka：它也提供了异步（async）处理模式，send()方法会返回Futrue对象，通过调用Futrue对象的get()方法，等待直到结果返回。

针对“情况 2”生产者缓冲区把消息弄丢了的解决思路：

方法1：可以用Redis的list、ES（可直接推送kafka）、本地文件充当缓冲区，来一条写一条，够数了再推。

针对“情况 3”MQ把消息弄丢了的解决思路：

1、必须开启持久化，就是消息写入mq后就会刷盘，就算mq自己挂掉了，恢复之后也会读取存储的数据，一般不会丢失，除非还没来得及持久化就挂掉了，这种情况很少，概率极低。

2、配合producer的confirm机制，只有消息被持久化或者大多数副本同步完成后，才返回ack；

3、kafka消费者丢失消息，跟rabbitmq处理机制是一样的，如：kafka

broker自己把消息给丢失了，生产者将消息发送给partion

leader，但是接收完，leader就宕机了，还没同步给follower，而马上另外的partion

follower被选举成了leader，这一块数据丢失了，怎么处理？必须将kafka的副本参数设置大于1，就是说至少有2个副本；kafka还需要设置副本leader至少感知不到一个follower挂掉了，才算leader挂掉了，才能重新选举；broker副本的follower全部同步成功时，才能算生产者写入消息队列写入成功【这样可以保证一条不丢】。

针对“情况 4”消费者拉取消息后，消息丢失的解决思路：

1、消费者丢失消息：一般是消费者打开了autoAck的机制，自动提交，但还在处理中，消费者系统宕机了，mq以为消费完了，造成了消费者丢失消息，怎么处理？

解决办法：关闭autoAck，每次处理完了再提交ack/offset，如果还没处理完就宕机了，长时间收不到ack，mq也会重新将这条消息分配给其他的消费者去处理。但是此时你需要自己保证消息的幂等性，因为如果消费者将消息已经消费了，但是返回给mq的确认信息失败了，mq以为你没消费，它会把消息发送给给另外的消费者，或者等该实例恢复后再次发送，引起消息的重复消费，此时，你可以将messageID做唯一键处理。

如果保证消息一条不丢，必须关闭消息过期清理的机制，否则消息积压会触发消息清理的机制，也会造成消息丢失。