0
点赞
收藏
分享

微信扫一扫

RabbitMQ高级特性(六):RabbitMQ之延迟队列


零、延迟队列与测试场景说明

延迟队列,即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费。

需求:

  1. 下单后,30分钟未支付,取消订单,回滚库存。
  2. 新用户注册成功7天后,发送短信问候。

实现方式:

  1. 定时器
  2. 延迟队列

RabbitMQ高级特性(六):RabbitMQ之延迟队列_java-rabbitmq

很可惜,在RabbitMQ中并未提供延迟队列功能。

但是可以使用:​TTL + 死信队列

RabbitMQ高级特性(六):RabbitMQ之延迟队列_rabbitmq_02

一、生产者工程

(1)RabbitMQ配置文件(rabbitmq.properties)

rabbitmq.host=192.168.116.161
rabbitmq.port=5672
rabbitmq.username=xiao
rabbitmq.password=xiao
rabbitmq.virtual-host=/myhost

(2)声明队列、交换机、交换机绑定队列(spring-rabbitmq-producer.xml)

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

<!--加载rabbitmq配置文件-->
<context:property-placeholder location="classpath:rabbitmq.properties"/>

<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"
publisher-confirms="true"
publisher-returns="true"/>
<!--定义管理交换机、队列-->
<rabbit:admin connection-factory="connectionFactory"/>

<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>


<!-- ========测试延迟队列==================================================================== -->
<!--
(1)声明死信交换机(order_exchange_dlx)和死信队列(order_queue_dlx)
(2)声明普通交换机(order_exchange)和队列(order_queue)
(3)绑定,设置普通队列过期时间为30分钟
-->

<!-- 1. 声明死信交换机(order_exchange_dlx)和死信队列(order_queue_dlx),两者绑定-->
<rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="order_exchange_dlx">
<rabbit:bindings>
<rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>

<!-- 2. 声明普通交换机(order_exchange)和普通队列(order_queue),两者绑定 -->
<rabbit:queue id="order_queue" name="order_queue">
<rabbit:queue-arguments>
<!-- 普通队列绑定死信交换机-->
<entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
<!-- 设置发送给死信交换机的routingkey-->
<entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
<!-- 设置队列的过期时间ttl(测试设置10秒)-->
<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="order_exchange">
<rabbit:bindings>
<rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>

</beans>

(2)测试类(发送消息到队列中)

package net.xiaof.test;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 测试延迟队列(场景:订单业务)
* @throws Exception
*/
@Test
public void testDelay() throws Exception {

//1.发送订单消息。 10秒后,消息进入死信队列,消费者端监听着死信队列,此时触发订单信息操作(取消订单/回滚库存等)
String msg = "订单信息:id=1,datetime=2020-12-27 01:40:21";
rabbitTemplate.convertAndSend("order_exchange","order.msg", msg);

//2.打印倒计时10秒(只为看延迟时间效果)
for (int i = 10; i > 0 ; i--) {
System.out.println(i+"...");
Thread.sleep(1000);
}

}

}

二、消费者工程

(1)RabbitMQ配置文件(rabbitmq.properties)

rabbitmq.host=192.168.116.161
rabbitmq.port=5672
rabbitmq.username=xiao
rabbitmq.password=xiao
rabbitmq.virtual-host=/myhost

(2)自定义Order(订单)死信队列监听器

package net.xiaof.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

/**
* @author zhangxh
* @Description: 消费端,定义Order(订单)死信队列监听器
* @date 2020-12-27
*/
@Component
public class OrderListener implements ChannelAwareMessageListener {

@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.out.println("OrderListener的onMessage方法执行了");

try {
//1.接收并转换消息
System.out.println(new String(message.getBody(),"UTF-8"));

//2.处理业务逻辑
System.out.println("处理业务逻辑...");
System.out.println("根据订单id查询其状态...");
System.out.println("判断状态是否为支付成功");
System.out.println("取消订单,回滚库存....");

//3.手动签收
channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();

//4.拒绝签收,不重回队列 requeue=false
/**
* long deliveryTag:该消息的index
* boolean multiple:是否批量。true:将一次性拒绝所有小于deliveryTag的消息。
* boolean requeue:被拒绝的是否重新入队列
*/
System.out.println("出现异常,拒绝签收...");
channel.basicNack(deliveryTag,true,false);
}

}

}

(3)开启手动确认模式、声明监听器(spring-rabbitmq-consumer.xml)

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!--加载配置文件-->
<context:property-placeholder location="classpath:rabbitmq.properties"/>

<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"/>

<!-- 组件扫描 -->
<context:component-scan base-package="net.xiaof.listener" />

<!--定义监听器容器(设置手动签收模式:acknowledge="manual",prefetch预处理数量)-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1" >
<!-- 引用自定义Order(订单)延迟队列监听器,监听死信队列order_queue_dlx -->
<rabbit:listener ref="orderListener" queue-names="order_queue_dlx"/>
</rabbit:listener-container>

</beans>

(4)测试类(运行死循环测试类)

package net.xiaof.test;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
* @author zhangxh
* @Description: 测试
* @date 2020-12-25
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {

/**
* 使用死循环开启消费端
*/
@Test
public void testConsumer() throws InterruptedException {
while (true){

}
}

}

三、测试方法

(1)运行消费者工程测试类(死循环运行),监听Order死信队列。

(2)运行生产者工程测试类,发送消息到普通队列(order_queue)中,10秒消息过期,自动路由到Order死信队列order_queue_dlx中,消费者端监听着死信队列,此时触发订单信息操作(取消订单/回滚库存等),运行结果如下图:

RabbitMQ高级特性(六):RabbitMQ之延迟队列_xml_03

四、延迟队列总结

(1)延迟队列 指消息进入队列后,可以被延迟一定时间,再进行消费。

(2)RabbitMQ没有提供延迟队列功能,但是可以使用 : TTL + DLX 来实现延迟队列效果。


举报

相关推荐

0 条评论