RabbitMQ延迟队列简单示例

Posted 2022-08-23 gjb724332682

简介

延迟队列存储的消息是不希望被消费者立刻拿到的，而是等待特定时间后，消费者才能拿到这个消息进行消费。使用场景比较多，例如订单限时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

执行完成