[大数据] RabbitMQ学习记录 - 死信队列（死信交换机）

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> RabbitMQ学习记录 - 死信队列（死信交换机） -> 正文阅读

[大数据]RabbitMQ学习记录 - 死信队列（死信交换机）

（内容均来自RabbitMQ官网：https://www.rabbitmq.com/dlx.html）

前面几篇学习了RabbitMQ的几种消息模型"HelloWorl"、"WorkQueue"、"Publish/Subscribe"、"Routing"、"Topic"，以及RabbitMQ中几个基本的组件：生产者、路由器、队列、消费者。

这篇学习下死信交换机，我觉得这么叫比较合理，因为RabbitMQ的官网也是这么叫的，名字也叫：Dead Letter Exchanges，简称：DLX。并不是网上大很多人说的死信队列。

什么情况下死信会投递到死信交换机去呢？

消息被消费者拒收了（消费者调用basic.reject或者basic.nack还有就是把requeue 的参数设置成false）
设置消息的TTL(x-message-ttl)
消息的数量超出了队列的最大长度限制(x-max-length)

情况一：消息被消费者拒收的情况产生死信

生产者代码

? ? ? ? 我们第一步准备演示情况1产生死信，消息的生产者这里做的事情主要是有：

创建普通交换机和普通队列
创建死信交换机和死信队列
在普通队列创建的时候携带参数（x-dead-letter-exchange：用于指定死信交换机的名称，x-dead-letter-routing-key：指定死信交换的routingKey、x-message-ttl指定消息过期时间、x-max-length队列最大消息容量等等）
发送消息

import com.booyue.tlh.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class DeadQueueProduce {

    //普通交换机名称、队列名称、routingKey
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String NORMAL_ROUTINGKEY = "normal_routing_key";

    //死信交换机名称、队列名称、routingKey
    public static final String DEAD_EXCHANGE = "dead_exchange";
    public static final String DEAD_QUEUE = "dead_queue";
    public static final String DEAD_ROUTINGKEY = "dead_routing_key";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getConnection().createChannel();
        //声明普通交换机交换机、死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT, true);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT, true);

        /**
         * 声明普通交换机时候需要的参数
         *  参数内容：可以指定死信交换机的名称、死信交换机的routingKey、队列的最大长度、队列中的消息过期时间等等
         */
        Map<String, Object> argsMap = new HashMap<>();
        //绑定死信队列名称
        argsMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //指定死信队列routingKey
        argsMap.put("x-dead-letter-routing-key", DEAD_ROUTINGKEY);
        //指定消息的过期时间为10000毫秒（10秒）
        //argsMap.put("x-message-ttl", 10000);
        //指定普通队列的最大长度
        // argsMap.put("x-max-length", 10);
        /**
         * 声明两个队列：普通交队列和死信队列
         */
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, argsMap);
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        /**
         * 将普通队列和普通交换机绑定、死信队列和死信交换机绑定
         */
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_ROUTINGKEY);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_ROUTINGKEY);

        /**
         * 发送消息
         */
        for (int i = 0; i < 10; i++) {
            String message = "info :" + i;
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_ROUTINGKEY, null, message.getBytes());
        }
    }
}

将生产者运行起来，我们在RabbitMQ的服务器上就能看到我们创建的普通交换机、普通队列（和死信交换机的绑定关系）、死信队列、死信交换机等等。

普通队列【途中看到的DLX和DLK用鼠标放上去，就能显示它绑定死信交换机和死信队列的routtingKey】和死信队列

普通交换机和死信交换机

普通消费者代码

????????这里需要改成手动确认的方式：autoAck=false，当消息是info :5的时候我们就拒收，让其进入到死信堆里中去，采用basicReject的方式拒收消息。其余与的消息就正常消费掉

import com.booyue.tlh.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

@Slf4j
public class NormalConsumer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String DEAD_QUEUE = "dead_queue";

    public static final String NORMAL_ROUTINGKEY = "normal_routing_key";
    public static final String DEAD_ROUTINGKEY = "dead_routing_key";

    public static void main(String[] args) throws IOException {
        //获取连接
        Connection connection = RabbitMQUtils.getConnection();
        //获取通道
        Channel channel = connection.createChannel();
        /**
         * 接受消息（当消息是info :5的时候我们就拒收，让其进入到死信堆里中去）
         *  这里需要改成手动确认的方式：autoAck=false
         *  采用basicReject的方式拒收消息
         */
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String ss = new String(message.getBody());
            if (ss.equals("info :5")) {
                //拒绝消息，不放回队列（该条消息就会到死信队列中）
                log.info("=====普通消费者拒收的信息：{}", ss);
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            } else {
                log.info("普通消费者收到的信息：{}", ss);
                //手动确认消息,不批量应答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        //修改成手动确认
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> {
        });
    }
}

把普通生产者和和普通消费者跑起来，看信息打印和RabbitMQ后台数据

? ? ? ? 看到信息打印，确实info：5的是消息被拒收了

20:43:33.971 [pool-1-thread-4] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :0
20:43:33.975 [pool-1-thread-4] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :1
20:43:33.975 [pool-1-thread-4] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :2
20:43:33.975 [pool-1-thread-4] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :3
20:43:33.975 [pool-1-thread-4] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :4
20:43:33.976 [pool-1-thread-5] INFO com.booyue.tlh.dead.NormalConsumer - =====普通消费者拒收的信息：info :5
20:43:33.976 [pool-1-thread-5] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :6
20:43:33.976 [pool-1-thread-5] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :7
20:43:33.976 [pool-1-thread-5] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :8
20:43:33.976 [pool-1-thread-5] INFO com.booyue.tlh.dead.NormalConsumer - 普通消费者收到的信息：info :9

RabbitMQ的服务器后台数据

? ? ? ? 在死信队列中确实都了一条数据，也确实是info：5

死信消息的消费者代码

import com.booyue.tlh.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

@Slf4j
public class DeadConsumer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String DEAD_QUEUE = "dead_queue";

    public static final String NORMAL_ROUTINGKEY = "normal_routing_key";
    public static final String DEAD_ROUTINGKEY = "dead_routing_key";

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

        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //接受消息
        DeliverCallback deliverCallback = (consumerTag, message) -> log.info("死信消费者受到的消息：{}",new String(message.getBody()));
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> {
        });
    }
}

死信消息的消费者获取到的信息打印

20:51:42.065 [pool-1-thread-4] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :5

情况二：设置消息的TTL(x-message-ttl)

? ? ? ? 在普通队列创建的时候，设置x-message-ttl，普通的消费者这次就不运行起来，模拟消息超时的情况。

生产者代码

? ? ? ? ?与第一种情况的差异就是把argsMap.put("x-message-ttl", 10000);注释打开了。默认超时时长为10000毫秒（10秒）

import com.booyue.tlh.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class DeadQueueProduce {

    //普通交换机名称、队列名称、routingKey
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String NORMAL_ROUTINGKEY = "normal_routing_key";

    //死信交换机名称、队列名称、routingKey
    public static final String DEAD_EXCHANGE = "dead_exchange";
    public static final String DEAD_QUEUE = "dead_queue";
    public static final String DEAD_ROUTINGKEY = "dead_routing_key";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getConnection().createChannel();
        //声明普通交换机交换机、死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT, true);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT, true);

        /**
         * 声明普通交换机时候需要的参数
         *  参数内容：可以指定死信交换机的名称、死信交换机的routingKey、队列的最大长度、队列中的消息过期时间等等
         */
        Map<String, Object> argsMap = new HashMap<>();
        //绑定死信队列名称
        argsMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //指定死信队列routingKey
        argsMap.put("x-dead-letter-routing-key", DEAD_ROUTINGKEY);
        //指定消息的过期时间为10000毫秒（10秒）
        argsMap.put("x-message-ttl", 10000);
        //指定普通队列的最大长度
        // argsMap.put("x-max-length", 10);
        /**
         * 声明两个队列：普通交队列和死信队列
         */
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, argsMap);
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        /**
         * 将普通队列和普通交换机绑定、死信队列和死信交换机绑定
         */
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_ROUTINGKEY);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_ROUTINGKEY);

        /**
         * 发送消息
         */
        for (int i = 0; i < 10; i++) {
            String message = "info :" + i;
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_ROUTINGKEY, null, message.getBytes());
        }
    }
}

死信消费者代码

? ? ? ?（和第一种情况时候的代码一样）

运行生产者和死信消费者，我们在RabbitMQ服务器后台能看到在普通队列里有10条消息积压

等待我们设置的超时时长，我们就能按到给条信息超时之后都会被死信消费者消费

21:01:20.530 [pool-1-thread-4] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :0
21:01:20.533 [pool-1-thread-4] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :1
21:01:20.533 [pool-1-thread-4] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :2
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :3
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :4
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :5
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :6
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :7
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :8
21:01:20.533 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :9

情况二：消息的数量超出了队列的最大长度限制(x-max-length)

????????在普通队列创建的时候，设置x-max-length（我们这里指定5），普通的消费者这次就不运行起来，让普通队列中的消息数量超过我呢设置的最大值5。与第一种情况的差异就是把argsMap.put("x-max-length", 5);注释打开了。默认队列的容量为5

生产者代码

import com.booyue.tlh.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class DeadQueueProduce {

    //普通交换机名称、队列名称、routingKey
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String NORMAL_QUEUE = "normal_queue";
    public static final String NORMAL_ROUTINGKEY = "normal_routing_key";

    //死信交换机名称、队列名称、routingKey
    public static final String DEAD_EXCHANGE = "dead_exchange";
    public static final String DEAD_QUEUE = "dead_queue";
    public static final String DEAD_ROUTINGKEY = "dead_routing_key";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getConnection().createChannel();
        //声明普通交换机交换机、死信交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT, true);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT, true);

        /**
         * 声明普通交换机时候需要的参数
         *  参数内容：可以指定死信交换机的名称、死信交换机的routingKey、队列的最大长度、队列中的消息过期时间等等
         */
        Map<String, Object> argsMap = new HashMap<>();
        //绑定死信队列名称
        argsMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //指定死信队列routingKey
        argsMap.put("x-dead-letter-routing-key", DEAD_ROUTINGKEY);
        //指定消息的过期时间为10000毫秒（10秒）
        //argsMap.put("x-message-ttl", 10000);
        //指定普通队列的最大长度
         argsMap.put("x-max-length", 5);
        /**
         * 声明两个队列：普通交队列和死信队列
         */
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, argsMap);
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);
        /**
         * 将普通队列和普通交换机绑定、死信队列和死信交换机绑定
         */
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_ROUTINGKEY);
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, DEAD_ROUTINGKEY);

        /**
         * 发送消息
         */
        for (int i = 0; i < 10; i++) {
            String message = "info :" + i;
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_ROUTINGKEY, null, message.getBytes());
        }
    }
}

死信消费者代码

? ? ? ?（和第一种情、第二中情况时候的代码一样）

运行生产者代码我们就能发现有五条消息在普通队列，另外5条在死信队列中（为了看效果我们先不运行死信队列，不然消息瞬间就被消费了）

运行死信消费者，看打印信息

? ? ? ? 只会消费死信队列里的信息，不会消费普通队列里面的信息

21:13:01.804 [pool-1-thread-4] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :0
21:13:01.809 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :1
21:13:01.809 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :2
21:13:01.809 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :3
21:13:01.809 [pool-1-thread-5] INFO com.booyue.tlh.dead.DeadConsumer - 死信消费者受到的消息：info :4

大数据最新文章

实现Kafka至少消费一次

亚马逊云科技：还在苦于ETL？Zero ETL的时代

初探MapReduce

【SpringBoot框架篇】32.基于注解+redis实现

Elasticsearch：如何减少 Elasticsearch 集

Go redis操作

Redis面试题

专题五 Redis高并发场景

基于GBase8s和Calcite的多数据源查询

Redis——底层数据结构原理

加:2021-07-09 17:33:49 更:2021-07-09 17:34:15

360图书馆购物三丰科技阅读网日历万年历 2024年5日历

-2024/5/4 18:59:00-

图片自动播放器
↓图片自动播放器↓

TxT小说阅读器
↓语音阅读,小说下载,古典文学↓

一键清除垃圾
↓轻轻一点,清除系统垃圾↓

图片批量下载器
↓批量下载图片,美女图库↓

网站联系: qq:121756557 email:121756557@qq.com IT数码