rabbitMQ死信队列快速编写记录

1.介绍

1.1 什么是死信队列

再rabbitMQ中，有两个重要的组件。exchange（交换机），queue（队列）。交换机用于路由消息，简单来说就是接收客户端传递的消息，转发到queue中。队列做的事情就是存储消息。
但消息并不会一只存储在队列中。当存在一下三种情况，消息就会死掉

• 队列存储不了过多的消息
• 消息本身存在过期时间

当遇到死掉的消息时，我们通常会将这些死信转发到新的交换机中，这个交换机就叫做死信交换机，而配合死信交换机存储信息的队列，叫做死信队列

1.2 死信队列有什么用

死信队列在构建延迟队列时，有巨大作用。比如用户购票订单，30min不支付就过期。在rabbitMQ中可以这样实现

• 1 存储购票信息到exchange-queue中
• 2 设置消息过期时间为30min
• 3 如果超过30min消息未被消费（消息过期，成为死信），存储死信队列，通知服务取消订单

2. 如何编码

2.1 架构分析

先上绑定架构图

整个流程中，出现了三方。在编写代码时，我们可以分三个大类，分别是producer,consumer1,consumer2。其消息传递顺序如下

• producer -> normal_exchange
• normal_queue -> consumer1
• dead_queue-> consumer2

我们可以在编写consumer1的时候，完成exchange和queue的创建与绑定

2.2 maven坐标

        <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.9.0</version></dependency>

2.3 工具类编写

rabbitMQ整体的编写流程如下

• 创建Connection工厂
• 配置工厂
• 创建链接
• 获取channel
• 通过channel创建exchange, queue, 关系绑定, 监听消息, 发送消息

因此，我们可以先创建工具类，帮我们获取channel，以此减少开发代码量

public class MQUtils {public static Channel getChannel() throws Exception {// 创建工厂链接ConnectionFactory factory = new ConnectionFactory();// 设置工厂factory.setHost("your_ip");factory.setUsername("your_username");factory.setPassword("your_password");factory.setVirtualHost("/"); // 基本都是/Connection connection = factory.newConnection();Channel channel = connection.createChannel();return channel;}
}

2.4 consumer1编写

public class Consumer1 {static String EXCHANGE_NAME = "normal_exchange";static String QUEUE_NAME = "normal_queue";static String DEAD_EXCHANGE_NAME = "dead_exchange";static String DEAD_QUEUE_NAME = "dead_queue";public static void main(String[] args) throws Exception{Channel channel = MQUtils.getChannel();// 普通交换机channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, true, null);// 普通队列// 配置死信交换机参数HashMap<String, Object> map = new HashMap<>();// 配置normal_queue连接的dead_exchangemap.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);// 设置normal_queue的消息过期时间map.put("x-message-ttl", 10000);// 设置路由到死信交换机的路由key: lisimap.put("x-dead-letter-routing-key", "lisi");// map.put("x-max-length", 6); // 设置队列最大长度channel.queueDeclare(QUEUE_NAME, false, false, true, map);// 绑定普通交换机和普通队列channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "zhangsan");// 死信交换机channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, true, false, null);// 死信队列channel.queueDeclare(DEAD_QUEUE_NAME, false, false, true, null);// 绑定死信交换机和死信队列channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, "lisi");// 监听channel.basicConsume(QUEUE_NAME, false, (consumerTag, message) -> {System.out.println("监听普通队列: " + new String(message.getBody()));}, consumerTag -> {});}
}

2.5 consumer2编写

public class Consumer2{static String DEAD_QUEUE_NAME = "dead_queue";public static void main(String[] args) throws Exception{Channel channel = MQUtils.getChannel();// 监听channel.basicConsume(DEAD_QUEUE_NAME, true, (consumerTag, message) -> {System.out.println("监听死信队列: " + new String(message.getBody()));}, consumerTag -> {});}
}

2.6 producer编写

public class Producer {static String EXCHANGE_NAME = "normal_exchange";public static void main(String[] args) throws Exception{Channel channel = MQUtils.getChannel();for (int i = 0; i < 10; i++) {String msg = i + "";channel.basicPublish(EXCHANGE_NAME, "zhangsan",null,msg.getBytes());}}
}

3.整合springboot

3.1 架构图

3.2 maven坐标

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.3.9.RELEASE</version></dependency>

3.3 构建配置类，创建exchange，queue，并绑定

package com.xhf.mq.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class TTLConfig {/*-----交换机名称-----*/private static final String EXCHANGE_NAME = "X";private static final String DEAD_EXCHANGE_NAME = "Y";/*-----队列名称-----*/private static final String QUEUE_NAME1 = "QA";private static final String QUEUE_NAME2 = "QB";private static final String DEAD_QUEUE_NAME = "QD";// 注册交换机@Bean("xExchange")public DirectExchange xExchange() {return ExchangeBuilder.directExchange(EXCHANGE_NAME).autoDelete().build();
//        return new DirectExchange(EXCHANGE_NAME);}// 注册死信交换机@Bean("deadExchange")public DirectExchange deadExchange() {return ExchangeBuilder.directExchange(DEAD_EXCHANGE_NAME).autoDelete().build();}// 注册队列QA@Bean("queueA")public Queue queueA() {return QueueBuilder.nonDurable(QUEUE_NAME1).withArgument("x-message-ttl", 10000).withArgument("x-dead-letter-exchange", DEAD_EXCHANGE_NAME).withArgument("x-dead-letter-routing-key", "YD").build();}// 注册队列QB@Bean("queueB")public Queue queueB() {return QueueBuilder.nonDurable(QUEUE_NAME2).withArgument("x-message-ttl", 40000).withArgument("x-dead-letter-exchange", DEAD_EXCHANGE_NAME).withArgument("x-dead-letter-routing-key", "YD").build();}// 注册队列QD@Bean("queueD")public Queue queueD() {return QueueBuilder.nonDurable(DEAD_QUEUE_NAME).build();}// 绑定普通交换机, 队列@Beanpublic Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") Exchange xExchange) {return BindingBuilder.bind(queueA).to(xExchange).with("XA").and(null);}@Beanpublic Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") Exchange xExchange) {return BindingBuilder.bind(queueB).to(xExchange).with("XB").and(null);}// 绑定死信交换机, 队列@Beanpublic Binding queueDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("deadExchange") Exchange deadExchange) {return BindingBuilder.bind(queueD).to(deadExchange).with("YD").and(null);}
}

3.4 编写生产者(controller的一个方法)

   @GetMapping("/sendToQA")public void sendToQA() {// 向X交换机发送消息, 消息通过"XA"路由到队列rabbitTemplate.convertAndSend("X", "XA", "hello".getBytes());}

3.5 编写消费者(一个类, 方法上加上@RabbitListenner，表明需要监听的queue)

@Component
public class Customer {@RabbitListener(queues = "QD")public void customer(Message message, Channel channel) {byte[] body = message.getBody();System.out.println(new String(body));}
}

4. 常见参数汇总

• x-dead-letter-exchange ：死信交换机名称
• x-message-ttl：消息time to live时间（过期时间）
• x-dead-letter-routing-key：死信交换机路由key
• x-max-length：队列最大长度