RabbitMQ消息100%不丢失？

消息消费流程&＃xff1a;

1.生产端发送消息到RabbitMQ;

2.RabbitMQ发送消息到消费端&＃xff1b;

3.消费端消费消息&＃xff1b;

以上3个步骤每个步骤都可能导致消息丢失&＃xff0c;消息丢失并不可怕&＃xff0c;可怕的是丢失了我们还不知道&＃xff0c;所以要有一系列措施来保证我们系统的可靠性。

生产端可靠性投递

一、事务消息机制

事务消息机制由于会严重降低性能&＃xff0c;所以一般不采用这种方法。

二、confirm消息确认机制

生产端投递的消息一旦投递到RabbitMQ后&＃xff0c;RabbitMQ就会发送一个确认消息给生产端&＃xff0c;让生产端知道我已经收到消息了&＃xff0c;否则这条消息就可能已经丢失了&＃xff0c;需要生产端重新发送消息了。

通过下面这句代码来开启确认模式&＃xff1a;

channel.confirmSelect();// 开启发送方确认模式

然后异步监听确认和未确认的消息&＃xff1a;

channel.addConfirmListener(new ConfirmListener() {//消息正确到达broker&＃64;Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {System.out.println("已收到消息");//做一些其他处理}//RabbitMQ因为自身内部错误导致消息丢失&＃xff0c;就会发送一条nack消息&＃64;Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {System.out.println("未确认消息&＃xff0c;标识&＃xff1a;" &＃43; deliveryTag);//做一些其他处理&＃xff0c;比如消息重发等}
});

三、消息持久化

RabbitMQ收到消息后将这个消息暂时存在了内存中&＃xff0c;如果RabbitMQ挂了&＃xff0c;那重启后数据就丢失了&＃xff0c;所以相关的数据应该持久化到硬盘中&＃xff0c;这样就算RabbitMQ重启后也可以到硬盘中取数据恢复。

message消息到达RabbitMQ后先是到exchange交换机中&＃xff0c;然后路由给queue队列&＃xff0c;最后发送给消费端。

 所有需要给exchange、queue和message都进行持久化&＃xff1a;

exchange持久化&＃xff1a;

//第三个参数true表示这个exchange持久化
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);

queue持久化&＃xff1a;

//第二个参数true表示这个queue持久化
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

message持久化&＃xff1a;

//第三个参数MessageProperties.PERSISTENT_TEXT_PLAIN表示这条消息持久化
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

四、消息入库

首先发送消息前先将消息保存到数据库中&＃xff0c;有一个状态字段status&＃61;0&＃xff0c;表示生产端将消息发送给了RabbitMQ但还没收到确认&＃xff1b;在生产端收到确认后将status设为1&＃xff0c;表示RabbitMQ已收到消息。这里有可能会出现上面说的两种情况&＃xff0c;所以生产端这边开一个定时器&＃xff0c;定时检索消息表&＃xff0c;将status&＃61;0并且超过固定时间后&＃xff08;可能消息刚发出去还没来得及确认这边定时器刚好检索到这条status&＃61;0的消息&＃xff0c;所以给个时间&＃xff09;还没收到确认的消息取出重发&＃xff08;第二种情况下这里会造成消息重复&＃xff0c;消费者端要做幂等性&＃xff09;&＃xff0c;可能重发还会失败&＃xff0c;所以可以做一个最大重发次数&＃xff0c;超过就做另外的处理。

 消费端消息不丢失

默认情况下&＃xff0c;以下3种情况会导致消息丢失&＃xff1a;

• 在RabbitMQ将消息发出后&＃xff0c;消费端还没接收到消息之前&＃xff0c;发生网络故障&＃xff0c;消费端与RabbitMQ断开连接&＃xff0c;此时消息会丢失&＃xff1b;

• 在RabbitMQ将消息发出后&＃xff0c;消费端还没接收到消息之前&＃xff0c;消费端挂了&＃xff0c;此时消息会丢失&＃xff1b;

• 消费端正确接收到消息&＃xff0c;但在处理消息的过程中发生异常或宕机了&＃xff0c;消息也会丢失。

 一、自动ack机制改为手动ack机制

因为RabbitMQ的自动ack机制&＃xff0c;即默认RabbitMQ在消息发出后就立即将这条消息删除&＃xff0c;而不管消费端是否接收到&＃xff0c;是否处理完&＃xff0c;导致消费端消息丢失时RabbitMQ自己又没有这条消息了。

消费端手动确认消息&＃xff1a;

DeliverCallback deliverCallback &＃61; (consumerTag, delivery) -> {try {//接收到消息&＃xff0c;做处理//手动确认channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (Exception e) {//出错处理&＃xff0c;这里可以让消息重回队列重新发送或直接丢弃消息}
};
//第二个参数autoAck设为false表示关闭自动确认机制&＃xff0c;需手动确认
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});

二、保证消息幂等性

让每个消息携带一个全局的唯一ID&＃xff0c;即可保证消息的幂等性&＃xff0c;具体消费过程为&＃xff1a;

1. 消费者获取到消息后先根据id去查询redis/db是否存在该消息
2. 如果不存在&＃xff0c;则正常消费&＃xff0c;消费完毕后写入redis/db
3. 如果存在&＃xff0c;则证明消息被消费过&＃xff0c;直接丢弃。

&＃64;Component
&＃64;RabbitListener(queuesToDeclare &＃61; &＃64;Queue(value &＃61; "javatrip", durable &＃61; "true"))
public class Consumer {&＃64;RabbitHandlerpublic void receiveMessage(Message message) throws Exception {Jedis jedis &＃61; new Jedis("localhost", 6379);String messageId &＃61; message.getMessageProperties().getMessageId();String msg &＃61; new String(message.getBody(),"UTF-8");System.out.println("接收到的消息为&＃xff1a;"&＃43;msg&＃43;"&＃61;&＃61;消息id为&＃xff1a;"&＃43;messageId);String messageIdRedis &＃61; jedis.get("messageId");if(messageId &＃61;&＃61; messageIdRedis){return;}JSONObject jsonObject &＃61; JSONObject.parseObject(msg);String email &＃61; jsonObject.getString("message");jedis.set("messageId",messageId);}
}