02_RabbitMQ消息丢失解决方案及死信队列

🕗 发布于 2024-09-23 01:33 java 开发语言

一、数据丢失

第一种：生产者弄丢了数据。生产者将数据发送到 RabbitMQ 的时候，可能数据就在半路给搞丢了，因为网络问题，都有可能。

第二种：RabbitMQ 弄丢了数据。MQ还没有持久化自己挂了。

第三种：消费端弄丢了数据。刚消费到，还没处理，结果进程挂了，比如重启了。

解决方案

消费者从队列中获取到数据，应答成功之后，队列认为消费者对消息处理完成，从队列中删除消息。

生产者

mq告知生产者，消息接收成功，没有成功，生产者可以继续重新发送消息

RabbitMQ有两种方式来解决这个问题：

通过AMQP提供的事务机制实现（同步，不推荐）；
使用发送者确认模式实现（confirm，异步）；

事务使用

事务的实现主要是对信道（Channel）的设置，主要的方法有三个：

channel.txSelect()声明启动事务模式；
channel.txCommit()提交事务；
channel.txRollback()回滚事务；

public class Product {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnUtils.getConn();
        Channel channel = conn.createChannel();

        String queueName = "my-que";
        // queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)
        channel.queueDeclare(queueName,false,false,false,null);

        String message = "hello rabbitmq";

        try{
            // 开启事务
            channel.txSelect();
            // 发布消息
            channel.basicPublish("",queueName,null,message.getBytes());
            // 提交事务
            channel.txCommit();
        }catch (Exception e){
            e.printStackTrace();
            // 回滚事务
            channel.txRollback();
        }
        
        channel.close();
        conn.close();
    }
}

Confirm发送方确认模式

Confirm发送方确认模式使用和事务类似，也是通过设置Channel进行发送方确认的。

Confirm的三种实现方式：

channel.waitForConfirms()普通发送方确认模式，单条数据；
channel.waitForConfirmsOrDie()批量确认模式，多条数据；
channel.addConfirmListener()异步监听发送方确认模式；

public class Product {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnUtils.getConn();
        // 创建信道：处理消息
        Channel channel = conn.createChannel();

        // 创建队列：存储消息
        String queueName = "my-que";
        // queueDeclare(String queue队列名, boolean durable持久化, boolean exclusive是否排外, boolean autoDelete自动删除,Map<String, Object> arguments队列其他属性信息)
        channel.queueDeclare(queueName, false, false, false, null);

        // 定义消息
        String message = "hello rabbitmq";

        // 开启发送方确认模式
        channel.confirmSelect();
        // basicPublish(交换器，序列名，参数信息，消息)
        channel.basicPublish("", queueName, null, message.getBytes());

        if(channel.waitForConfirms()){
            System.out.println("发送成功");
        }

        channel.close();
        conn.close();
    }
}

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

for (int i = 0; i < 10; i++) {
    channel.basicPublish("", queueName, null, message.getBytes());
}

// 直到所有信息都发布，只要有一个未确认就会IOException
channel.waitForConfirmsOrDie(); 
System.out.println("全部执行完成");

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

for (int i = 0; i < 10; i++) {
    channel.basicPublish("", queueName, null, message.getBytes());
}

//异步监听确认和未确认的消息
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("未确认消息，标识：" + deliveryTag);
    }
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
    }
});

System.out.println("执行操作...");

MQ

rabbitmq做持久化，持久化主要分为：

交换器的持久化

// 参数1 exchange ：交换器名
// 参数2 type ：交换器类型
// 参数3 durable ：是否持久化
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);

队列持久化

// 参数1 queue ：队列名
// 参数2 durable ：是否持久化
// 参数3 exclusive ：仅创建者可以使用的私有队列，断开后自动删除
// 参数4 autoDelete : 当所有消费客户端连接断开后，是否自动删除队列
// 参数5 arguments
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

消息持久化

// 参数1 exchange ：交换器
// 参数2 routingKey ： 路由键
// 参数3 props ： 消息的其他参数,其中 MessageProperties.PERSISTENT_TEXT_PLAIN 表示持久化
// 参数4 body ： 消息体
channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

消费者

关闭自动应答，改为手动应答，处理完成，MQ中再做删除

        // 消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("deliveryTag:" + envelope.getDeliveryTag());
                Thread.sleep(1000);
                System.out.println("获取到的消息内容:" + new String(body));
                // 手动应答
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };

        // false:关闭自动应答
        channel.basicConsume(queueName, false, consumer);

二、死信队列

消息过期会自动删除
消费者立刻接收消息进行处理

消费者立刻接收到数据处理，需求队列得数据不立刻被消费者处理，而是希望等待一段时间才被处理

死信队列：是延迟队列得一种实现方式，不是一种特殊得队列，死信交换机+消息有效期

交换机的类型：fanout，direct，topic，header
队列：不区分类型，也是一个普通得队列
死信交换机：将一个交换机设置为某一个队列的死信交换机，当前队列过期的消息会发送给交换机
消息：有效期

消息传递顺序

生产者发布消息 --> 交换机(转发消息) --> 队列(接收存储消息) --> 消费者(接收处理消息)

队列转发消息：

正常消息：转给消费者处理；
过期消息：默认删除，也可以通过配置（死信交换机）转发给交换机；

交换机的消息来源：

生产者发布的（普通得生产者）
其他队列到期删除的消息自动进入队列（队列需要绑定交换机，队列发布）

队列指的内容：

消息来源都是从交换机发送得

死信队列：交换机B作为队列A的死信交换机，接收队列A中被删除的消息，实现队列A中的消息延迟处理的效果

死信队列DLX（dead-letter-exchange），利用DLX，当消息在一个队列中变成死信 (dead message) 之后，它能被重新投递到另一个exchange上，这个exchange就是DLX，似乎叫死信交换机更加贴切，当死信投递到这个exchange后，我们也可以用一个queue来绑定该exchange，该exchange就可以根据路由规则把这个死信路由到这个queue了，同样可以创建消费端去进行消费，以便对死信进行相应的处理。

延时队列：实际是不存在直接可用的延时队列，可通过死信消息和死信队列来实现延时队列的功能。RabbitMQ中的所有队列，消息存进来，立刻分发给对应的消费者进行处理。

死信交换机： DLX 全称（Dead-Letter-Exchange）。其实它是个普通的交换机，但它是设置在队列上某个参数的值对应的交换机。

死信队列：如果某个队列上存在参数：x-dead-letter-exchange，当这个队列里的消息变成死信消息（dead message）后会被重新Pushlish到 x-dead-letter-exchange 所对应参数值的交换机上，跟这个交换机所绑定的队列就是死信队列。

为什么使用延迟队列

订单五分钟内支付：未支付订单超时

以前：1.用户做支付时，判断是否超时，修改订单状态；（不调用）

2.定时任务，每一秒刷新一次；（频繁刷新）

延时队列的应用场景很多，在我的项目开发中也涉及到很多，例如：订单五分钟未支付自动取消、订单准备超时30分钟推送提醒给门店、订单完成后两小时推送评价邀请给用户等等，这些间隔指定时间后的操作都可以使用延时队列。

消息变成死信的情况

死信消息：

消息被拒绝(basic.reject / basic.nack)，并且requeue = false（不重回队列）。
消息TTL过期。
队列达到最大长度。

过期消息：RabbitMq 有两种设置消息过期的方式：

创建队列时通过 x-message-ttl 参数指定该队列消息的过期时间，这种队列里的消息过期时间全部相同。
生产者Pushlish消息时，通过设置消息的 expiration 参数指定过期时间，每个消息的过期时间都不一样。

　　如果两者同时使用，过期时间按照小的一方为准，两种方式设置的时间都是毫秒。

Queue参数信息

name: 队列的名称；
actualName: 队列的真实名称，默认用name参数，如果name为空，则根据规则生成一个；
durable: 是否持久化；
exclusive: 是否独享、排外的；
autoDelete: 是否自动删除；
arguments：队列的其他属性参数，有如下可选项，可参看图2的arguments：
- x-message-ttl：消息的过期时间，单位：毫秒；
- x-expires：队列过期时间，队列在多长时间未被访问将被删除，单位：毫秒；
- x-max-length：队列最大长度，超过该最大值，则将从队列头部开始删除消息；
- x-max-length-bytes：队列消息内容占用最大空间，受限于内存大小，超过该阈值则从队列头部开始删除消息；
- x-overflow：设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。有效值是drop-head、reject-publish或reject-publish-dlx。仲裁队列类型仅支持drop-head；
- x-dead-letter-exchange：死信交换器名称，过期或被删除（因队列长度超长或因空间超出阈值）的消息可指定发送到该交换器中；
- x-dead-letter-routing-key：死信消息路由键，在消息发送到死信交换器时会使用该路由键，如果不设置，则使用消息的原来的路由键值
- x-single-active-consumer：表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，其他被忽略，false时消息循环分发给所有消费者(默认false)
- x-max-priority：队列要支持的最大优先级数;如果未设置，队列将不支持消息优先级；
- x-queue-mode（Lazy mode）：将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用;如果未设置，队列将保留内存缓存以尽可能快地传递消息；
- x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。

交换机参数信息

类型

FanoutExchange ：扇形交换机
DirectExchange ：直连交换机 routing模式用的交换机
TopicExchange ：主题模式交换机

参数信息

name：交换机名称
durable：是否把交换机持久化到磁盘上
autoDelete ：是否自动删除交换机用法有点类似上面的队列中的autoDelete 只有所有队列都和交换机接触订阅，说白了就是所有绑定到交换机上的队列不再需要改交换机，他就该死了,有点残酷
artuments ：额外参数

实现

生产者

发布消息，设置消息过期时间

public class Producer {

    // default exchange
    private static String exchange = "test_dlx_exchange";
    // default exchange 的路由规则： routingKey（test） 将匹配同名的 queue(test)
    private static String routingKey = "dlx.abc";

    public static void main(String[] args) throws IOException, TimeoutException {

        // 1 创建Connection
        Connection connection = ConnUtils.getConn();

        // 2 创建Channel
        Channel channel = connection.createChannel();

        // 3 发送消息
        for (int i = 0; i < 5; i++) {
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    // deliveryMode:1（nopersistent）非持久化，2（persistent）持久化
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    // expiration:消息的失效时间
                    .expiration("50000")
                    .build();
            String msg = "RabbitMQ: dlx message" + i;
            channel.basicPublish(exchange , routingKey , properties , msg.getBytes());
        }

        // 5 关闭连接
        channel.close();
        connection.close();
    }
}

正常消费者

绑定死信队列，当消息变为死信消息时，会自动存入私信队列

public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnUtils.getConn();

        Channel channel = connection.createChannel();

        // 定义死信队的Exchange
        String dlxExchange = "dlx.exchange";
        channel.exchangeDeclare(dlxExchange, "topic");

        // 死信队列名
        String dlxQueue = "dlx.queue";
        channel.queueDeclare(dlxQueue, true, false, false, null);

        // # 表示所有的key都可以路由到s死信队列
        String dlxRoutingKey = "#";
        // 绑定死信队列和exchange
        channel.queueBind(dlxQueue, dlxExchange, dlxRoutingKey, null);

        // 定义正常的消费者j监听队列
        // 申明exchange
        String exchangeName = "test_dlx_exchange";
        channel.exchangeDeclare(exchangeName, "topic");

        // 申明队列
        Map<String, Object> arguments = new HashMap<>();
        // 设置死信队列,arguments要设置到申明的队列上
        arguments.put("x-dead-letter-exchange", dlxExchange);
        String queueName = "test_dlx_queue";
        channel.queueDeclare(queueName, true, false, false, arguments);

        // 队列绑定到exchange
        String routingKey = "dlx.#";
        channel.queueBind(queueName, exchangeName, routingKey);

        channel.basicQos(1);

        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("---消费者-- " + new String(message.getBody(), "UTF-8"));

            }
        };

        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("---消费者--：cancelCallback ");
            }
        };

        // callback: 消费者会调函数，处理发送过来的消息。
        // cancelCallback: 消费者取消订阅时的回调方法。
        // 消费消息,autoAck一定要设为false，手工ack
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback);
    }
}

死信消费者

处理死信队列中得信息

// DLXConusmer,监听消费死信队列中的消息
public class DLXConusmer {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnUtils.getConn();

        Channel channel = connection.createChannel();

        String queueName = "dlx.queue";
        String exchangeName = "dlx.exchange";
        String routingKey = "#";

        // 申明exchange
        channel.exchangeDeclare(exchangeName, "topic");
        // 申明队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 队列绑定到exchange
        channel.queueBind(queueName, exchangeName, routingKey, null);

        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {
                    System.out.println("---死信队列消费者---");
                    System.out.println(new String(message.getBody(), "UTF-8"));
                } finally {
                    // consumer手动 ack 给broker
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }
            }
        };
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("---消费者：cancelCallback");
            }
        };

        // 消费消息,autoAck一定要设置为false
        channel.basicConsume(queueName, false, deliverCallback, cancelCallback);
    }
}

原文地址：https://blog.csdn.net/dys000000/article/details/142445702

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：django项目——图片上传到阿里云OSS对象存储
下一篇：电磁兼容相关概念

【网络安全】Cookie SameSite属性
严格限制可能会影响用户体验，例如，当用户点击 GitHub 链接时，无法保持登录状态，因为此时会丢失 GitHub 的 Cookie。当浏览器加载该图片时，它会向 A.com 发送带有 Cookie
阅读更多2024-11-16
【C语言】实现二维数组按行排序
【C语言】实现二维数组按行排序
阅读更多2024-11-16
NLP开发常见问题
NLP开发常见问题
阅读更多2024-11-16
在spring boot工程中使用Filter时，@WebFilter 注解不生效的问题分析和解决方案
如果要使用@WebFilter进行过滤路径的配置，则需要在启动类中添加@ServletComponentScan注解启用Servlet容器扫描@WebFilter注解，并删除Filter类中的@Com
阅读更多2024-11-16
天天学编程Day16
在 C++ 中，常量表达式（Constant Expression）是指在编译期就能计算出结果的表达式。其结果在程序运行过程中是不可改变的，并且这个结果必须是编译时常量，例如整数常量、枚举常量、用常量
阅读更多2024-11-16
消失的数字（c语言实现）
方法一我们使用异或操作符实现。介绍一下^操作符性质就是首先0^a=a；a^a=0;然后异或操作符还满足交换律。比如a ^ b ^ c ^a= a ^ a ^b ^c;我们可以通过这样的性质实现这个代
阅读更多2024-11-16
机器学习——期末复习重点题归纳
机器学习期末复习
阅读更多2024-11-16
opencv kdtree & pcl kdtree 效率对比
由于项目中以一个环节需要使用kdtree ,对性能要求比较严苛，所以看看那个kdtree效率高一些。对比了opencv和pcl。
阅读更多2024-11-16
Qt 5.6.3 手动配置 mingw 环境
手动设置 qt mingw 编译环境
阅读更多2024-11-16
IntelliJ IDEA 2024.3（Ultimate Edition）免费化教学
一款由 JetBrains 公司开发的集成开发环境 (Integrated Development Environment, IDE)，主要用于 Java 语言的软件开发，但它也支持其他多种编程语言和
阅读更多2024-11-16