• RabbitMQ
  • RabbitMQ消息丢失3种情况
    • 生产者弄丢了数据
    • RabbitMQ弄丢了数据
    • 消费者弄丢了数据
  • RabbitMQ处理方案
  • Kafka
    • 消费者弄丢了数据
    • Kafka弄丢了数据

RabbitMQ

RabbitMQ消息丢失3种情况

生产者弄丢了数据

生产者将数据发送到RabbitMQ时候，可能数据半路丢失，因为网络问题之类的。

此时可以用RabbitMQ提供的事务功能，就是生产者发送数据之前开启RabbitMQ事务channel.txSelect，然后发送消息，如果消息没有被RabbitMQ成功接收到，那么生产者会报异常，此时可以回滚事务channel.txRollback，然后重试发送消息；如果收到了消息，那么可以提交事务channel.txCommit。

// 开启事务
channel.txSelect
try {
    // 这里发送消息
} catch (Exception e) {
    channel.txRollback

    // 这里再次重发这条消息
}

// 提交事务
channel.txCommit

然而事务机制的影响，非常消耗性能

可以开启confirm模式，在生产者设置这个模式，每次写的消息都会分配一个唯一的id，如果写入了RabbitMQ中，会回传一个ack消息，告诉你这个消息ok；如果RabbitMQ没能处理这个消息，会回调一个nack接口，通知消息接收失败，可以重试。可以结合这个机制在自己的内存维护每个消息id的状态，如果超过一定时间没接收到这个消息的回调，可以重试。

事务与confirm机制的不同在于，事务机制是同步的，提交一个事务会阻塞在那边，但是confirm是异步的，发送这个消息可以再继续发送下一个消息，然后RabbitMQ接收之后会异步回调接口通知你这个消息接收到了。所以一般都是使用confirm机制。

RabbitMQ弄丢了数据

RabbitMQ 还没持久化，自己就挂了，可能导致少量数据丢失

• 创建queue的时候将其设为持久化 可以保证RabbitMQ持久化queue的元数据，但是不会持久化queue里面的数据
• 发送消息的时候将消息的deliveryMode设置为2 就是将消息设置为持久化，RabbitMQ就会将消息持久化到磁盘

必须同时设置这2个持久化才能保证

持久化可以跟生产者那边的 confirm 机制配合起来，只有消息被持久化到磁盘之后，才会通知生产者 ack 了，所以哪怕是在持久化到磁盘之前，RabbitMQ 挂了，数据丢了，生产者收不到 ack ，你也是可以自己重发的

消费者弄丢了数据

因为你消费的时候，刚消费到，还没处理，结果进程挂了，比如重启了，RabbitMQ 认为你都消费了，这数据就丢了

可以用RabbitMQ提供的ack机制，也就是关闭RabbitMQ的自动ack，通过一个api调用，每次代码确保处理完的时候，在程序里ack一次。如果你还没处理完，就没有ack了，这个时候RabbitMQ认为你还没处理完，就会把这个消费分给其他的消费者去处理

RabbitMQ处理方案

Kafka

消费者弄丢了数据

消费了这个消息，消费者那边自动提交了offset，让Kafka以为已经消费好了这个消息，但实际上才刚准备处理，还没处理，自己就挂了，此时这条消息就丢失。

关闭自动提交offset，处理完之后手动提交offset，但是可能会出现重复消费，比如刚处理完但是还没提交offset，只能自己保证幂等性

Kafka弄丢了数据

Kafka某个broker宕机，重新选举partition的leader。如果其他的follower还有些数据没有同步，此时leader挂了，然后开始选举某个follower为leader就丢失了一些数据。

一般设置以下4个参数

• 给topic设置replication.factor 参数：这个值必须大于 1，要求每个 partition 必须有至少 2 个副本。
• 在 Kafka 服务端设置 min.insync.replicas 参数：这个值必须大于 1，这个是要求一个 leader 至少感知到有至少一个 follower 还跟自己保持联系，没掉队，这样才能确保 leader 挂了还有一个 follower 吧。
• 在 producer 端设置 acks=all ：这个是要求每条数据，必须是写入所有 replica 之后，才能认为是写成功了。
• 在 producer 端设置 retries=MAX （很大很大很大的一个值，无限次重试的意思）：这个是要求一旦写入失败，就无限重试，卡在这里了。