RabbitMQ 死信队列
https://www.rabbitmq.com/dlx.html
DLX 即 Dead-Letter-Exchange 也叫做死信交换机。
死信队列是指队列上的消息变成死信后，能够后发送到另外一个交换机，这个交换机 就是 DLX 。
一般有几种情况会变成死信：
- 消息被拒绝（ Basic.reject 或者 basic.nack）并且设置 requeue 参数为 false
- 消息 过期 设置了 message TTL
- 队列达到最大的长度
死信交换机是正常的交换机，能够在任何队列上被指定。其实死信交换机和一般的交换机没啥区别，只是添加了死信交换机的属性。如果队列上存在死信， RabbitMq 会将死信消息投递到设置的 DLX 上去 ，然后被路由到一个队列上，这个队列，就是死信队列。
流程如下：
生产者：
import java.util.HashMap;
import java.util.Map;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setHost("xxxx");
connectionFactory.setUsername("xxx");
connectionFactory.setPassword("xxx");
connectionFactory.setPort(5672);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchangeName = "normal_exchange";
String routingkey = "dlx.dlx";
String msg = "test dlx message";
String queueName = "normal_queueName";
Map<String,Object>map =new HashMap<>();
//注意：x-dead-letter-exchange 这个key是固定这样写的，value是你自定义的。
map.put("x-dead-letter-exchange","exchange.dlx");
//注意：arguments需要声明在队列上，声明在交换机上是不会起作用的。
channel.queueDeclare(queueName, true, false, false, map);
channel.exchangeDeclare(exchangeName,"direct",true,false,null);
channel.queueBind(queueName,exchangeName,"dlx.dlx");
for (int i = 0; i < 3; i++) {
// deliveryMode=2 持久化，expiration 消息有效时间
AMQP.BasicProperties properties=new AMQP.BasicProperties().builder()
.deliveryMode(2)
.contentEncoding("utf-8")
.expiration("7000")
.build();
channel.basicPublish(exchangeName, routingkey, true, properties, msg.getBytes());
}
}
}
消费者：
import java.io.IOException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
public class Consumer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setHost("xxxx");
connectionFactory.setPort(5672);
connectionFactory.setUsername("xxxx");
connectionFactory.setPassword("xxxx");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//死信交换机声明
channel.exchangeDeclare("exchange.dlx","topic",true,false,null);
//死信队列声明
channel.queueDeclare("queue.dlx",true,false,false,null);
//routingkey指定为#就行，表示只要路由到死信队列的都接收
channel.queueBind("queue.dlx","exchange.dlx","#");
channel.basicConsume("queue.dlx", false, "myConsumer Tag", new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
String routingKey = envelope.getRoutingKey();
String convernType = properties.getContentType();
long deliveryTag = envelope.getDeliveryTag();
System.out.println("routingKey:"+routingKey+",convernType:"+convernType+",deliveryTag:"+deliveryTag+",Msg body:"+new String(body));
channel.basicAck(deliveryTag, false);
}
});
}
}
上面的代码可以看到：
消息通过正常交换机 normal_exchange 到达了正常队列 normal_queue。
map.put("x-dead-letter-exchange","exchange.dlx");
//注意：arguments需要声明在队列上，声明在交换机上是不会起作用的。
channel.queueDeclare(queueName, true, false, false, map);
正常的队列 normal_queue 声明了下面参数下设置了一个 x-dead-letter-exchange 当消息过期时，将消息发送到死信交换机 exchange.dlx
死信交换机下面绑定了一个队列 queue.dlx
channel.exchangeDeclare("exchange.dlx","topic",true,false,null);
//死信队列声明
channel.queueDeclare("queue.dlx",true,false,false,null);
//routingkey指定为#就行，表示只要路由到死信队列的都接收
channel.queueBind("queue.dlx","exchange.dlx","#");
最后将消息发送到了死信队列上，消费者，消费死信队列 queue.dlx 上的消息即可
channel.basicConsume("queue.dlx", false, "myConsumer Tag", new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
String routingKey = envelope.getRoutingKey();
String convernType = properties.getContentType();
long deliveryTag = envelope.getDeliveryTag();
System.out.println("routingKey:"+routingKey+",convernType:"+convernType+",deliveryTag:"+deliveryTag+",Msg body:"+new String(body));
channel.basicAck(deliveryTag, false);
}
});
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2019-08-22，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读