首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >18-RabbitMQ高级特性-死信队列

18-RabbitMQ高级特性-死信队列

作者头像
彼岸舞
发布2022-10-06 08:40:57
2160
发布2022-10-06 08:40:57
举报

死信队列

死信队列: DLX, Dead-Letter-Exchange

  • 利用DLX, 当消息在一个队列中变成死信(dead message)之后, 它能被重新publish到另一个Exchange, 这个Exchange就是DLX
  • DLX也是一个正常的Exchange, 和一般的Exchange没有区别, 他能在任何的队列上被指定, 实际上就是设置某个队列的属性
  • 当这个队列中有死信时,RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去, 进而被路由到另一个队列
  • 可以监听这个队列的消息做相应的处理, 这个特征可以弥补RabbitMQ3.0以前支持的immediate参数功能

消息变成死信的情况

  • 消息被拒绝(basic.reject/basic.nack), 并且requeue=false(关闭重回队列)
  • 消息TTL过期
  • 队列达到最大长度

死信队列的设置

  • 首先需要设置死信队列的Exchange和Queue, 然后进行绑定
    • Exchange: dlx.exchange
    • Queue: dlx.queue
    • RoutingKey: #
  • 然后进行正常的声明交换机, 队列, 绑定, 只不过需要在队列上添加一个参数即可
    • arguments.put("x-dead-letter-exchange","dlx.exchange");
  • 这样消息在过期, requeue=false, 队列在达到最大长度时, 消息就直接路由到死信队列了!

死信队列代码实现

消费者

package com.dance.redis.mq.rabbit.dlx;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
 
public class Receiver4DLXtExchange {
 
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        // 声明正常的 exchange queue 路由规则
        String queueName = "test_dlx_queue";
        String exchangeName = "test_dlx_exchange";
        String routingKey = "group.*";
        RabbitMQHelper.exchangeDeclare(channel,exchangeName,RabbitMQHelper.EXCHANGE_TYPE_TOPIC);

        //    注意在这里要加一个特殊的属性arguments: x-dead-letter-exchange
        Map<String, Object> arguments = new HashMap<>();
        // 指定死信队列
        arguments.put("x-dead-letter-exchange", "dlx.exchange");
        // 指定死信队列的路由规则
        arguments.put("x-dead-letter-routing-key", "dlx.*");
        RabbitMQHelper.queueDeclare(channel,queueName,true,arguments);
        channel.queueBind(queueName, exchangeName, routingKey);

        // 声明死信队列
        // dlx declare:
        RabbitMQHelper.exchangeDeclare(channel,"dlx.exchange",RabbitMQHelper.EXCHANGE_TYPE_TOPIC);
        channel.queueDeclare("dlx.queue", false, false, false, null);
        channel.queueBind("dlx.queue", "dlx.exchange", "#");
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException {
                // 拒绝并且不重回队列, 这样就会进入死信队列
                channel.basicReject(envelope.getDeliveryTag(),false);
            }
        };
        //    参数:队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, false, consumer);
        //等待回调函数执行完毕之后,关闭资源。
        TimeUnit.SECONDS.sleep(50);
        channel.close();
        RabbitMQHelper.closeConnection();
    }
}

生产者

package com.dance.redis.mq.rabbit.dlx;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.HashMap;
import java.util.Map;

public class Sender4DLXExchange {


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        String exchangeName = "test_dlx_exchange";
        String routingKey = "group.dlx";
        Map<String, Object> headers = new HashMap<>();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .deliveryMode(2)
                .contentEncoding("UTF-8")
                .headers(headers).build();
        String msg = "Hello World RabbitMQ 4 DLX Exchange Message ... ";
        channel.basicPublish(exchangeName, routingKey, props, msg.getBytes());
    }

}

死信队列测试

启动消费者

启动生产者

查看消费者

应为直接拒绝了, 所以没有消费, 查看控制台

已经被加入到死信队列中了, 为啥是3呢, 应为我之前测试了两次, 这个时候, 如果是写业务的话, 就可以通过消费死信队列的消息, 完成消费失败的, 或者过期的补偿了~

我这里只是用了拒绝策略, TTL过期和队列满, 都会进入, 可以自己试一下

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2022-10-04,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

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

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 死信队列
    • 死信队列: DLX, Dead-Letter-Exchange
      • 消息变成死信的情况
        • 死信队列的设置
          • 死信队列代码实现
            • 死信队列测试
            领券
            问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档