Rabbitmq延迟队列学习
1概述
1.1延迟队列
延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。
应用场景:
下单后，30分钟未支付，取消订单，回滚库存。
30分钟内字符,则从MQ消息队列中修改消息状态half-->可发送。
30分钟到, 未支付,则取消订单,回滚库存,删除消息。
在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列 组合实现延迟队列的效果。
2代码示例
2.1生产端
package com.kikop.workpattern.topic.producer;
import com.kikop.common.utils.DateUtil;
import com.kikop.common.utils.RabbitMQUtils;
import com.kikop.config.RabbitMQConfig;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
/**
* @author kikop
* @version 1.0
* @project Name: myrabbitmqdemo
* @file Name: WeatherBureau
* @desc 功能描述 消费消息
* @date 2020/9/6
* @time 18:46
* @by IDE: IntelliJ IDEA
*/
public class DelayedQueueProducer {
/**
* 单独为某条消息设置超时
*
* @throws IOException
* @throws TimeoutException
*/
private static void createDelayQueueTest() throws IOException, TimeoutException {
Connection connection = RabbitMQUtils.getConnection();
final Channel channel = connection.createChannel();
// 定义死信交换机
boolean durable_dlx_exchange = true;
channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_DLX_TOPIC, BuiltinExchangeType.TOPIC, durable_dlx_exchange);
// 1.过期队列及属性设置
Map<String, Object> arguments = new HashMap<String, Object>();
// 1.1.设置整个队列过期时间
// arguments.put("x-message-ttl", 10000);
// 1.2.设置队列的私信参数(队列过期的后处理,移到死信交换机中)
arguments.put("x-dead-letter-exchange", RabbitMQConfig.EXCHANGE_DLX_TOPIC);
arguments.put("x-dead-letter-routing-key", "rk_mydlx");
// 1.3.创建持久化队列
boolean durable_queue = true;
channel.queueDeclare(RabbitMQConfig.QUEUE_DELAY_TOPIC_PRIVATE, durable_queue, false, false, arguments);
// 2.创建持久化转发代理
boolean durable_exchange = true;
channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, BuiltinExchangeType.TOPIC, durable_exchange);
String routingKey = "mydelay";
channel.queueBind(RabbitMQConfig.QUEUE_DELAY_TOPIC_PRIVATE, RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, routingKey);
// 3.设置消息持久化
String message = "购票超时,订单编号:000001,锁定的库存编号为:000110,即将执行回滚,(增加库存,作废订单)操作!";
int durable_msg = 2; // 设置消息是否持久化，1： 非持久化 2：持久化
AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder();
properties.deliveryMode(durable_msg); // 设置消息是否持久化，1： 非持久化 2：持久化
// 4.发送 mock half消息
properties.expiration(3 * 60 * 1000 + ""); // 过期时间3分钟
System.out.println(DateUtil.getCurrentDateStr());
channel.basicPublish(RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, routingKey,
properties.build(), message.getBytes("UTF-8"));
channel.close();
connection.close();
}
public static void main(String[] args) throws Exception {
createDelayQueueTest();
return;
}
}
2.2消费端
package com.kikop.workpattern.topic.consumer;
import com.kikop.common.utils.DateUtil;
import com.kikop.common.utils.RabbitMQUtils;
import com.kikop.config.RabbitMQConfig;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author kikop
* @version 1.0
* @project Name: myrabbitmqdemo
* @file Name: BiaDu
* @desc 功能描述 消费消息
* @date 2020/9/6
* @time 18:46
* @by IDE: IntelliJ IDEA
*/
public class DelayedQueueConsumer {
public static void main(String[] args) throws IOException, TimeoutException {
delayConsumerTest();
return;
}
private static void delayConsumerTest() throws IOException, TimeoutException {
Connection connection = RabbitMQUtils.getConnection();
final Channel channel = connection.createChannel();
// begin_配置通道队列(可选,可在生产端完成)
// 定义通道的非持久化队列
channel.queueDeclare(RabbitMQConfig.QUEUE_DLX_TOPIC, false, false, false, null);
// 定义需要消费的私信队列 QUEUE_DLX_TOPIC,从关联的死信交换机中抓取
// 增加现有的一个转发代理,代理 EXCHANGE_DLX_TOPIC 必须先存着
// queueBind用于将队列与交换机绑定
// 参数1：队列名
// 参数2：交互机名
// 参数三：路由key
channel.queueBind(RabbitMQConfig.QUEUE_DLX_TOPIC, RabbitMQConfig.EXCHANGE_DLX_TOPIC, "rk_mydlx");
// end_配置通道队列(可选)
channel.basicQos(1);
// 只消费死信队列里面的消息
channel.basicConsume(RabbitMQConfig.QUEUE_DLX_TOPIC, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(DateUtil.getCurrentDateStr() + "@信息：" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}
2.3测试
// 9.延迟队列(TTL+DLX)
// 实际上是私有的
public static final String EXCHANGE_DELAY_TOPIC_PRIVATE = "ex_delay_topic_private";
public static final String QUEUE_DELAY_TOPIC_PRIVATE = "queue_delay_topic_private";
public static final String EXCHANGE_DLX_TOPIC = "ex_dlx_topic";
public static final String QUEUE_DLX_TOPIC = "queue_dlx_topic";
生产端:2021-03-28 23:02:47
消费端:2021-03-28 23:05:47@信息：购票超时,订单编号:000001,锁定的库存编号为:000110,即将执行回滚,(增加库存,作废订单)操作!
参考
1RabbitMQ消息最终一致性解决方案
https://blog.csdn.net/weixin_43466542/article/details/101676944
2.RabbitMq(十一) 死信交换机DLX介绍及使用
https://blog.csdn.net/liuhenghui5201/article/details/107538775
