这是我参与11月更文挑战的第10天，活动详情查看：2021最后一次更文挑战
现在主流的消息队列产品都提供了非常完善的消息可靠性保证机制，完全可以做到在消息传递过程中，即使发生网络中断或者硬件故障，也能确保消息的可靠传递，不丢消息。
一.检测消息丢失的方法
我们可以利用消息队列的有序性来验证是否有消息丢失。在Producer端，我们给每个发出的消息附加一个连续递增的序号，然后在Consumer端来检查这个序号的连续性。如果没有消息丢失，Consumer 收到消息的序号必然是连续递增的，或者说收到的消息，其中的序号必然是上一条消息的序号 +1。如果检测到序号不连续，那就是丢消息了。还可以通过缺失的序号来确定丢失的是哪条消息，方便进一步排查原因。
二.确保消息可靠传递
一条消息从生产到消费完成这个过程，可以划分三个阶段，分别是“生产阶段”，“存储阶段”，“消费阶段”。
生产阶段: 在这个阶段，从消息在 Producer 创建出来，经过网络传输发送到 Broker 端。
存储阶段: 在这个阶段，消息在 Broker 端存储，如果是集群，消息会在这个阶段被复制到其他的副本上。
消费阶段: 在这个阶段，Consumer 从 Broker 上拉取消息，经过网络传输发送到 Consumer 上。
1. 生产阶段
在生产阶段，消息队列通过最常用的请求确认机制，来保证消息的可靠传递:当你的代码调 用发消息方法时，消息队列的客户端会把消息发送到 Broker，Broker 收到消息后，会给客 户端返回一个确认响应，表明消息已经收到了。客户端收到响应后，完成了一次正常消息的 发送。
只要Producer收到了Broker的确认响应，就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后，会自动重试，如果重试再失败，就会以返回值或者 异常的方式告知用户。
正确处理返回值或者捕获异常，就可以保证这个阶段的消息不会丢失。以 Kafka 为例，我们看一下如何可靠地发送消息:
同步发送时，只要注意捕获异常即可。
try {
RecordMetadata metadata = producer.send(record).get();
System.out.println(" 消息发送成功。");
} catch (Throwable e) {
System.out.println(" 消息发送失败!");
System.out.println(e);
}
复制代码异步发送时，则需要在回调方法里进行检查。
producer.send(record,(metadata,exception) -> {
if(metadata != null) {
System.out.println("消息发送成功")；
}else {
System.out.println("消息发送失败")；
}
}
复制代码2.存储阶段
在存储阶段正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。
如果对消息的可靠性要求非常高，可以通过配置 Broker 参数来避免因为宕机丢消息。
对于单个节点的 Broker，需要配置 Broker 参数，在收到消息后，将消息写入磁盘后再给 Producer 返回确认响应，这样即使发生宕机，由于消息已经被写入磁盘，就不会丢失消息，恢复后还可以继续消费。例如，在 RocketMQ 中，需要将刷盘方式flushDiskType 配置为 SYNC_FLUSH 同步刷盘。
如果是 Broker 是由多个节点组成的集群，需要将 Broker 集群配置成:至少将消息发送到 2 个以上的节点，再给客户端回复发送确认响应。这样当某个 Broker 宕机时，其他的 Broker 可以替代宕机的 Broker，也不会发生消息丢失。后面我会专门安排一节课，来讲解 在集群模式下，消息队列是如何通过消息复制来确保消息的可靠性的。
3.消费阶段
消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从 Broker 拉取消 息后，执行用户的消费业务逻辑，成功后，才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在 网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。
在编写消费代码时需要注意的是，不要在收到消息后就立即发送消费确认，而是应该在执 行完所有消费业务逻辑之后，再发送消费确认。
三.小结
-
在生产阶段，需要捕获消息发送的错误，并重发消息。
-
在存储阶段，可以通过配置刷盘和复制相关的参数，让消息写入到多个副本的磁盘上，来确保消息不会因为某个 Broker 宕机或者磁盘损坏而丢失。
-
在消费阶段，需要在处理完全部消费业务逻辑之后，再发送消费确认。
四.RabbitMQ的可靠性传输
1.1 RabbitMQ出现消息丢失的情况及其解决办法
1.1.1 生产者没有成功把消息发送到MQ
a、丢失的原因： 因为网络传输的不稳定性，当生产者在向MQ发送消息的过程中，MQ没有成功接收到消息，但是生产者却以为MQ成功接收到了消息，不会再次重复发送该消息，从而导致消息的丢失。
b、解决办法： 事务机制和confirm机制
1.事务机制
RabbitMQ 提供了事务功能，生产者发送数据之前开启 RabbitMQ 事务channel.txSelect，然后发送消息，如果消息没有成功被 RabbitMQ 接收到，那么生产者会收到异常报错，此时就可以回滚事务channel.txRollback，然后重试发送消息；如果收到了消息，那么可以提交事务channel.txCommit。伪代码如下：
// 开启事务
channel.txSelect
try {
// 这里发送消息
} catch (Exception e) {
channel.txRollback
// 这里再次重发这条消息
}
// 提交事务
channel.txCommit
复制代码2.confirm机制：
RabbitMQ可以开启 confirm 模式，在生产者那里设置开启 confirm 模式之后，生产者每次写的消息都会分配一个唯一的 id，如果消息成功写入 RabbitMQ 中，RabbitMQ 会给生产者回传一个 ack 消息，告诉你说这个消息 ok 了。如果 RabbitMQ 没能处理这个消息，会回调你的一个 nack 接口，告诉你这个消息接收失败，生产者可以发送。而且你可以结合这个机制自己在内存里维护每个消息 id 的状态，如果超过一定时间还没接收到这个消息的回调，那么可以重发。
注意：RabbitMQ的事务机制是同步的，很耗型能，会降低RabbitMQ的吞吐量。confirm机制是异步的，生成者发送完一个消息之后，不需要等待RabbitMQ的回调，就可以发送下一个消息，当RabbitMQ成功接收到消息之后会自动异步的回调生产者的一个接口返回成功与否的消息。
1.1.2 RabbitMQ接收到消息之后丢失了消息
a、丢失的原因：
RabbitMQ接收到生产者发送过来的消息，是存在内存中的，如果没有被消费完，此时RabbitMQ宕机了，那么再次启动的时候，原来内存中的那些消息都丢失了。
b、解决办法：
开启RabbitMQ的持久化。当生产者把消息成功写入RabbitMQ之后，RabbitMQ就把消息持久化到磁盘。结合上面的说到的confirm机制，只有当消息成功持久化磁盘之后，才会回调生产者的接口返回ack消息，否则都算失败，生产者会重新发送。存入磁盘的消息不会丢失，就算RabbitMQ挂掉了，重启之后，他会读取磁盘中的消息，不会导致消息的丢失。
c、持久化的配置：
第一点是创建 queue 的时候将其设置为持久化，这样就可以保证 RabbitMQ 持久化 queue 的元数据，但是它是不会持久化 queue 里的数据的。
第二个是发送消息的时候将消息的 deliveryMode 设置为 2，就是将消息设置为持久化的，此时 RabbitMQ 就会将消息持久化到磁盘上去。
注意：持久化要起作用必须同时设置这两个持久化才行，RabbitMQ 哪怕是挂了，再次重启，也会从磁盘上重启恢复 queue，恢复这个 queue 里的数据。
1.1.3 消费者弄丢了消息
a、丢失的原因：
如果RabbitMQ成功的把消息发送给了消费者，那么RabbitMQ的ack机制会自动的返回成功，表明发送消息成功，下次就不会发送这个消息。但如果就在此时，消费者还没处理完该消息，然后宕机了，那么这个消息就丢失了。
b、解决的办法：
简单来说，就是必须关闭 RabbitMQ 的自动 ack，可以通过一个 api 来调用就行，然后每次在自己代码里确保处理完的时候，再在程序里 ack 一把。这样的话，如果你还没处理完，不就没有 ack了？那 RabbitMQ 就认为你还没处理完，这个时候 RabbitMQ 会把这个消费分配给别的 consumer 去处理，消息是不会丢的。
文章参考：
blog.csdn.net/zengdongwen…
感谢李玥老师指导。
本章节并不是完整的，后续还会继续更新，修改。
近期评论