标签：efault eth 集群 使用场景 简介 dir 版本 消费者 queue
简介
延迟队列存储的消息是不希望被消费者立刻拿到的，而是等待特定时间后，消费者才能拿到这个消息进行消费。使用场景比较多，例如订单限时30分钟内支付，否则取消，再如分布式环境中每隔一段时间重复执行某操作。
下面举一个简单的例子，例子大概意思是分别在首次发送消息后的10秒、40秒、100秒后重新读取到消息。为了直观，不使用RabbitMQ其他多余的特性。
准备工作
在Centos7下安装RabbitMQ，版本为3.6.12单机版本（非集群），IP是127.0.0.1，端口是15672，使用web管理页面或者rabbitmqctl提前准备好相关的用户、exchange和queue。
用户有producer（密码同用户名）、consumer（密码同用户名）。
Default exchange是RabbitMQ预定义的，名称为空字符串，自动绑定到每个queue，类型为direct，routingKey等于queue的名称。
三个死信交换器（队列中的消息过期后会被发送到该队列的死信交换器）10sDeadLetterExchange、30sDeadLetterExchange、60sDeadLetterExchange。
三个死信队列，10sDeadLetterQueue、30sDeadLetterQueue、60sDeadLetterQueue分别与三个死信交换器绑定。
三个队列，10sDelayQueue、30sDelayQueue、60sDelayQueue，分别设置过期时间为10秒、30秒、60秒，并设置对应的死信交换器。
示例
使用java代码创建一个生产者和三个消费者。生产者往Default exchange发送消息，routingKey为10sDelayQueue。三个消费者分别订阅三个死信队列。先启动三个消费者，再启动生产者。消费者会持续订阅死信队列，需要手动关闭连接。
Rabbitmq10sDeadLetterQueueConsumer消费者代码如下：
import com.rabbitmq.client.*;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
public class Rabbitmq10sDeadLetterQueueConsumer {
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("consumer");
connectionFactory.setPassword("consumer");
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(15672);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.basicQos(64);
channel.basicConsume("10sDeadLetterQueue", false, UUID.randomUUID().toString(),
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("第一次重试时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println("consumerTag:" + consumerTag);
System.out.println("envelope:" + envelope.toString());
System.out.println("basicProperties:" + properties.toString());
System.out.println("message:" + new String(body, "utf-8"));
channel.basicAck(envelope.getDeliveryTag(), false);
channel.basicPublish("", "30sDelayQueue",
new AMQP.BasicProperties().builder().deliveryMode(2).build(),
body);
}
});
}
}
Rabbitmq30sDeadLetterQueueConsumer消费者代码如下：
import com.rabbitmq.client.*;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
public class Rabbitmq30sDeadLetterQueueConsumer {
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("consumer");
connectionFactory.setPassword("consumer");
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(15672);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.basicQos(64);
channel.basicConsume("30sDeadLetterQueue", false, UUID.randomUUID().toString(),
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("第二次重试时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println("consumerTag:" + consumerTag);
System.out.println("envelope:" + envelope.toString());
System.out.println("basicProperties:" + properties.toString());
System.out.println("message:" + new String(body, "utf-8"));
channel.basicAck(envelope.getDeliveryTag(), false);
channel.basicPublish("", "60sDelayQueue",
new AMQP.BasicProperties().builder().deliveryMode(2).build(),
body);
}
});
}
}
Rabbitmq60sDeadLetterQueueConsumer消费者代码如下：
import com.rabbitmq.client.*;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
public class Rabbitmq60sDeadLetterQueueConsumer {
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("consumer");
connectionFactory.setPassword("consumer");
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(15672);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.basicQos(64);
channel.basicConsume("60sDeadLetterQueue", false, UUID.randomUUID().toString(),
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("第三次重试时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println("consumerTag:" + consumerTag);
System.out.println("envelope:" + envelope.toString());
System.out.println("basicProperties:" + properties.toString());
System.out.println("message:" + new String(body, "utf-8"));
channel.basicAck(envelope.getDeliveryTag(), false);
System.out.println("执行完成");
}
});
}
}
生产者代码如下：
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
public class RabbitmqDelayQueueProducer {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("producer");
connectionFactory.setPassword("producer");
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(15672);
connection = connectionFactory.newConnection();
channel = connection.createChannel();
channel.basicPublish("", "10sDelayQueue",
new AMQP.BasicProperties().builder().deliveryMode(2).build(),
("测试延迟队列" + UUID.randomUUID().toString()).getBytes("utf-8"));
System.out.println("发送时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
} finally {
if (channel != null) {
channel.close();
}
if (connection != null) {
connection.close();
}
}
}
}
生产者打印：
发送时间：2019-08-25 10:08:29
Rabbitmq10sDeadLetterQueueConsumer消费者打印：
第一次重试时间：2019-08-25 10:08:39
consumerTag:83787685-28a9-4ae8-b2bc-b89b90f14b68
envelope:Envelope(deliveryTag=1, redeliver=false, exchange=10sDeadLetterExchange, routingKey=10sDelayQueue)
basicProperties:#contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=expired, count=1, exchange=, time=Sun Aug 25 10:08:37 CST 2019, routing-keys=[10sDelayQueue], queue=10sDelayQueue}]}, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
message:测试延迟队列f862c616-a99a-4b65-9482-d74d5aee0814
Rabbitmq30sDeadLetterQueueConsumer消费者打印：
第二次重试时间：2019-08-25 10:09:09
consumerTag:27bb0e0e-07fe-49b6-8c0f-06fbc32cd784
envelope:Envelope(deliveryTag=1, redeliver=false, exchange=30sDeadLetterExchange, routingKey=30sDelayQueue)
basicProperties:#contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=expired, count=1, exchange=, time=Sun Aug 25 10:09:07 CST 2019, routing-keys=[30sDelayQueue], queue=30sDelayQueue}]}, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
message:测试延迟队列f862c616-a99a-4b65-9482-d74d5aee0814
Rabbitmq60sDeadLetterQueueConsumer消费者打印：
第三次重试时间：2019-08-25 10:10:09
consumerTag:8bfdc795-443d-4940-bf88-1e6f14e7b530
envelope:Envelope(deliveryTag=1, redeliver=false, exchange=60sDeadLetterExchange, routingKey=60sDelayQueue)
basicProperties:#contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=expired, count=1, exchange=, time=Sun Aug 25 10:10:07 CST 2019, routing-keys=[60sDelayQueue], queue=60sDelayQueue}]}, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
message:测试延迟队列f862c616-a99a-4b65-9482-d74d5aee0814
执行完成
RabbitMQ延迟队列简单示例
标签：efault eth 集群 使用场景 简介 dir 版本 消费者 queue
原文地址：https://www.cnblogs.com/gjb724332682/p/11407296.html
