以下几种情况导致消息变为死信:
对于RabbitMQ 来说,DLX 是一个非常有用的特性。它可以处理异常情况下,消息不能够被消费者正确消费(消费者调用了Basic.Nack 或者Basic.Reject)而被置入死信队列中的情况,后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况,进而可以改善和优化系统。
package com.lagou.rabbitmq.demo;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 DeadProducer {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setUri("amqp://root:123456@192.168.80.121:5672/%2f");Connection connection = factory.newConnection();Channel channel = connection.createChannel();Map arguments = new HashMap<>();// 设置队列中消息TTLarguments.put("x-message-ttl", 10000);// 设置该队列所关联的死信交换器(当队列消息TTL到期后依然没有消费,则加入死信队列)arguments.put("x-dead-letter-exchange", "ex.dlx");// 设置该队列所关联的死信交换器的routingKey,如果没有特殊指定,使用原队列的routingKeyarguments.put("x-dead-letter-routing-key", "key.dlx");// 定义一个正常业务的交换器channel.exchangeDeclare("ex.biz", "direct", true);// 定义一个正常队列channel.queueDeclare("queue.biz", true, false, false, arguments);// 正常队列绑定channel.queueBind("queue.biz", "ex.biz", "key.biz");// 定义一个死信交换器(也是一个普通的交换器)channel.exchangeDeclare("ex.dlx", "direct", true);// 定义一个死信队列channel.queueDeclare("queue.dlx", true, false, false, null);// 死信队列和死信交换器channel.queueBind("queue.dlx", "ex.dlx", "key.dlx");channel.basicPublish("ex.biz","key.biz",null,"orderid.8484494".getBytes());channel.close();connection.close();}
}
org.springframework.boot spring-boot-starter-amqp org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.junit.vintage junit-vintage-engine org.springframework.amqp spring-rabbit-test test
spring.application.name=dlx
spring.rabbitmq.host=node1
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=root
spring.rabbitmq.password=123456
spring.rabbitmq.port=5672
package com.lagou.rabbitmq.demo;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class RabbitmqDemo {public static void main(String[] args) {SpringApplication.run(RabbitmqDemo08.class, args);}
}
package com.lagou.rabbitmq.demo.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class RabbitConfig {@Beanpublic Queue queue() {Map props = new HashMap<>();// 消息的生存时间 10sprops.put("x-message-ttl", 10000);// 设置该队列所关联的死信交换器(当队列消息TTL到期后依然没有消费,则加入死信队列)props.put("x-dead-letter-exchange", "ex.go.dlx");// 设置该队列所关联的死信交换器的routingKey,如果没有特殊指定,使用原队列的routingKeyprops.put("x-dead-letter-routing-key", "go.dlx");Queue queue = new Queue("q.go", true, false, false, props);return queue;}@Beanpublic Queue queueDlx() {Queue queue = new Queue("q.go.dlx", true, false, false);return queue;}@Beanpublic Exchange exchange() {DirectExchange exchange = new DirectExchange("ex.go", true, false, null);return exchange;}/*** 死信交换器** @return*/@Beanpublic Exchange exchangeDlx() {DirectExchange exchange = new DirectExchange("ex.go.dlx", true, false, null);return exchange;}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(exchange()).with("go").noargs();}/*** 死信交换器绑定死信队列** @return*/@Beanpublic Binding bindingDlx() {return BindingBuilder.bind(queueDlx()).to(exchangeDlx()).with("go.dlx").noargs();}
}
package com.lagou.rabbitmq.demo.controller;import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class GoController {@Autowiredprivate AmqpTemplate rabbitTemplate;@RequestMapping("/go")public String distributeGo() {rabbitTemplate.convertAndSend("ex.go", "go", "送单到石景山x小区,请在10秒内接受任务");return "任务已经下发,等待送单。。。";}@RequestMapping("/notgo")public String getAccumulatedTask() {String notGo = (String) rabbitTemplate.receiveAndConvert("q.go.dlx");return notGo;}
}