前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >RabbitMQ如何保证消息的可靠性

RabbitMQ如何保证消息的可靠性

原创
作者头像
王二蛋
修改2024-02-07 17:05:03
1630
修改2024-02-07 17:05:03
举报
文章被收录于专栏:消息中间件消息中间件

可靠性分析

RabbitMQ如何保证消息的可靠?如RabbitMQ基础概念中的架构模型

RabbitMQ架构模型
RabbitMQ架构模型

可以看到一条消息的传递过程:

  1. 发布者和RabbitMQ建立连接发送消息至交换机。
  2. 交换机和队列绑定,将消息路由到队列中。
  3. 消费者和RabbitMQ建立连接指定某个队列的消息进行消费。

在这过程中以下几个环节可能会丢失消息:

  1. 发布者到交换机环节。
  2. 交换机到队列环节。
  3. 队列到消费者环节。

如下图

RabbitMQ丢失消息的环节
RabbitMQ丢失消息的环节

可靠性方案

所以要保证消息的可靠性需要做到以下几点:

  1. 发布者需确认交换机接收到消息。
  2. 发布者需确认队列接收到消息。
  3. 保证队列及其中的数据持久化。
  4. 保证消费者的正常消费。

如何做到以上几点?RabbitMQ为了适应各个场景的使用,以上的功能需要开发者按照定义自行设置实现。


可靠性实现

以下是Java整合RabbitMQ的实现,参考Java整合RabbitMQ实现生产消费(7种通讯方式)

确认Exchange接收到消息

构建channel时添加确认监听机制,当消息未发送至交换机时做补偿措施。

代码语言:java
复制
 channel.addConfirmListener((sequenceNumber, multiple) -> {
            System.out.println("消息成功发送到交换机");
        }, (sequenceNumber, multiple) -> {
            System.err.println("消息未发送到交换机,补偿操作。");
        });

确认Queue接收到消息

  • 构建channel时添加return监听机制,当消息未路由至队列时做补偿措施。
代码语言:java
复制
channel.addReturnListener((replyCode, replyText, exchange, routingKey, basicProperties, body) -> {
            System.err.format("消息 %s 未路由到指定队列: %s, replyText: %s,replyCode: %d%n", body, routingKey, replyText, replyCode);
        });
  • 发布消息时设置消息mandatory为true,开启return机制。
代码语言:java
复制
 channel.basicPublish("", "",true, "", "");

保证Queue及其数据持久化

  • 构建队列时持久化队列。
代码语言:java
复制
 //构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除","其他参数")
  channel.queueDeclare("", true, false, false, null);
  • 发布消息时设置消息持久化属性。
代码语言:java
复制
//设置消息持久化
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().deliveryMode(2).build();
channel.basicPublish("", "",true, basicProperties, "");     

保证消费者的正常消费

当消息持久化至队列时已经保证了消息的可靠投递,为保证消息的正常消费,需要解决重复消费和消息丢失问题。

重复消费问题

  1. 业务处理完成,但是ack失败,消息被扔进队列,导致重复消费。
  2. 业务处理过程中,进程宕机,恢复进程后消费未ACK的消息导致重复消费。

针对第一个场景的解决方案:设置手动ACK,并且业务处理和ack操作在一个事务中。

代码语言:java
复制
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            //消息处理后手动ACK
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
// ack为false
channel.basicConsume("", false, deliverCallback, consumerTag -> {
});

针对第二个场景的解决方案:发布消息时设置业务唯一标识,在消费后进行存储,如果有相同标识前来消费直接拒绝即可(具体业务具体分析)。设置业务唯一标识方式:

代码语言:java
复制
String correlationId = UUID.randomUUID().toString();
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(correlationId).build();
channel.basicPublish("", "", basicProperties, "");     

消息丢失问题

在以下场景中可能会发生消息丢失问题:

  1. 业务处理失败,但是ack成功,导致消息丢失。

解决方案:设置手动ACK,并且业务处理和ack操作在一个事务中。

总结

RabbitMQ 本身可以保证消息的可靠性,但是需要开发者去了解整体的流程,并且根据实际情况去自行保证。

我正在参与2024腾讯技术创作特训营第五期有奖征文,快来和我瓜分大奖!

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 可靠性分析
  • 可靠性方案
  • 可靠性实现
    • 确认Exchange接收到消息
      • 确认Queue接收到消息
        • 保证Queue及其数据持久化
          • 保证消费者的正常消费
            • 重复消费问题
            • 消息丢失问题
        • 总结
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
        http://www.vxiaotou.com