使用RabbitMQ完成消息延迟推送
背景
本周做了一个需求，大体是这样的，一个用户领取了一条信息（公用的（领取后变成私有的）），但是如果他在设定的时间内没有使用这条信息，那么这条信息将会被释放掉，由于本人之前没有做过类似的需求，遂找前辈取经，得到了两种解决方案，一种是自己写定时器，每隔一小时或者多长时间跑一次满足释放条件的数据，然后释放掉，另一种就是使用RabbitMQ的延迟队列，在听了延迟队列的作用后果断选择此方法
什么是延迟队列
用过RabbitMQ的同学都知道，一般的队列，消息一旦入队了之后就会被消费者马上消费，而延迟队列就是说，消息进入后会根据延迟时间长短来消费，也就是超时才会被消费
实现思路
RabbitMQ允许我们为消息或者队列设置TTL（time to live），也就是过期时间。TTL表明了一条消息可在队列中存活的最大时间，单位为毫秒。也就是说，当某条消息被设置了TTL或者当某条消息进入了设置了TTL的队列时，这条消息会在经过TTL毫秒后“死亡”，成为Dead Letter。
还有另一个特性：如果队列设置了Dead Letter Exchange（DLX），那么这些Dead Letter就会被重新publish到Dead Letter Exchange，通过Dead Letter Exchange路由到其他队列。
根据这两个特性我们不难想到满足当前需求的延迟队列的实现方案，也就是TTL和DLX特性结合在一起，流程图大概是这样的：
创建MQ相关配置
- 创建一个测试的Virtual Host，我这里叫LxTestMqHost
- 创建对应的DLX和延迟重试的Exchange(我这里叫test_per_queue_ttl_exchange)
- 创建实际消费队列
- 创建缓冲队列并绑定DLX和消费队列
- 创建消费失败的缓冲队列并绑定test_per_queue_ttl_exchange和消费队列还有失效时间
- 绑定DLX和消费队列
配置MQ代码
package com.yqn.crm.mq.config;
import org.Springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class QueueConfig extends BaseConfig {
/**
* DLX
*/
public final static String DELAY_EXCHANGE_NAME = "test_delay_exchange";
/**
* 路由到test_delay_queue_per_queue_ttl的exchange
*/
public final static String PER_QUEUE_TTL_EXCHANGE_NAME = "test_per_queue_ttl_exchange";
/**
* 发送到该队列的message会在一段时间后过期进入到test_delay_process_queue
* 每个message可以控制自己的失效时间
*/
public final static String DELAY_QUEUE_PER_MESSAGE_TTL_NAME = "test_delay_queue_per_message_ttl";
/**
* 发送到该队列的message会在一段时间后过期进入到test_delay_process_queue
* 队列里所有的message都有统一的失效时间
*/
public final static String DELAY_QUEUE_PER_QUEUE_TTL_NAME = "test_delay_queue_per_queue_ttl";
public final static int QUEUE_EXPIRATION = 5000;
/**
* message失效后进入的队列，也就是实际的消费队列
*/
public final static String DELAY_PROCESS_QUEUE_NAME = "test_delay_process_queue";
/**
* 创建DLX exchange
*
* @return
*/
@Bean
DirectExchange delayExchange() {
return new DirectExchange(DELAY_EXCHANGE_NAME);
}
/**
* 创建test_per_queue_ttl_exchange
*
* @return
*/
@Bean
DirectExchange perQueueTTLExchange() {
return new DirectExchange(PER_QUEUE_TTL_EXCHANGE_NAME);
}
/**
* 创建test_delay_queue_per_message_ttl队列
*
* @return
*/
@Bean
Queue delayQueuePerMessageTTL() {
return QueueBuilder.durable(DELAY_QUEUE_PER_MESSAGE_TTL_NAME)
.withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME) // DLX，dead letter发送到的exchange
.withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME) // dead letter携带的routing key
.build();
}
/**
* 创建test_delay_queue_per_queue_ttl队列
*
* @return
*/
@Bean
Queue delayQueuePerQueueTTL() {
return QueueBuilder.durable(DELAY_QUEUE_PER_QUEUE_TTL_NAME)
.withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME) // DLX
.withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME) // dead letter携带的routing key
.withArgument("x-message-ttl", QUEUE_EXPIRATION) // 设置队列的过期时间
.build();
}
/**
* 创建test_delay_process_queue队列，也就是实际消费队列
*
* @return
*/
@Bean
Queue delayProcessQueue() {
return QueueBuilder.durable(DELAY_PROCESS_QUEUE_NAME)
.build();
}
/**
* 将DLX绑定到实际消费队列
*
* @param delayProcessQueue
* @param delayExchange
* @return
*/
@Bean
Binding dlxBinding(Queue delayProcessQueue, DirectExchange delayExchange) {
return BindingBuilder.bind(delayProcessQueue)
.to(delayExchange)
.with(DELAY_PROCESS_QUEUE_NAME);
}
/**
* 将per_queue_ttl_exchange绑定到delay_queue_per_queue_ttl队列
*
* @param delayQueuePerQueueTTL
* @param perQueueTTLExchange
* @return
*/
@Bean
Binding queueTTLBinding(Queue delayQueuePerQueueTTL, DirectExchange perQueueTTLExchange) {
return BindingBuilder.bind(delayQueuePerQueueTTL)
.to(perQueueTTLExchange)
.with(DELAY_QUEUE_PER_QUEUE_TTL_NAME);
}
@Bean(name = "lxTestMqReleaseConnectionFactory")
public ConnectionFactory connectionElaneFactory() {
String virtualHost="LxTestMqHost";
return getConnectionFactory(host, port, username, password, virtualHost);
}
@Bean(name = "lxTestMqRabbitListenerContainerFactory")
public SimpleRabbitListenerContainerFactory lxTestMqRabbitListenerContainerFactory(@Qualifier("lxTestMqReleaseConnectionFactory") ConnectionFactory connectionFactory, RabbitProperties config) {
return getFactory(connectionFactory, config);
}
@Bean(name = "lxTestMqReleaseTemplate")
public RabbitTemplate orderSummaryAmqpTemplate(@Qualifier("lxTestMqReleaseConnectionFactory") ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
return template;
}
}
测试MQ代码
package com.yqn.crm.mapper.crm;
import com.yqn.crm.mq.config.QueueConfig;
import com.yqn.crm.mq.receiver.ExpirationMessagePostProceSSOr;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
/**
* @author 刘鑫
* @description
* @since 2019-03-24 17:24
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class Mqtestww {
@Autowired
@Qualifier("lxTestMqReleaseTemplate")
private AmqpTemplate rabbitTemplate;
@Test
public void testDelayQueuePerMessageTTL() throws Exception {
for (int i = 1; i <= 3; i++) {
long expiration = i * 5000;//超时时间设置
rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_MESSAGE_TTL_NAME,
(Object) ("测试手动设置超时时间" + expiration), new ExpirationMessagePostProcessor(expiration));//可以将想要的参数放到message中，参数可以是json类型的所以可以将对象传上去
}
}
@Test
public void testDelayQueuePerQueueTTL() throws InterruptedException {
for (int i = 1; i <= 3; i++) {
rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_QUEUE_TTL_NAME,
"测试自动超时时间 " + QueueConfig.QUEUE_EXPIRATION);
}
}
/*/**
* @author 凑合看
* @description 测试接收消息
* @since 2019-03-24 19:05
* @param [msg, channel, message]
* @return void
**/
@RabbitListener(bindings =
@QueueBinding(value = @Queue(name = QueueConfig.DELAY_PROCESS_QUEUE_NAME),
exchange = @Exchange(name = QueueConfig.DELAY_EXCHANGE_NAME)),
containerFactory = "lxTestMqRabbitListenerContainerFactory")
public void consumer(@Payload String msg, Channel channel, Message message) throws Exception {
Throwable throwable = null;
try {
String test = msg;//将msg消息打印下来，如果推送消息的时候message中放的是一个对象，也可以使用json来转成想要的对象
System.out.println(test);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
// 如果发生了异常，则将该消息重定向到缓冲队列，会在一定延迟之后自动重做
channel.basicPublish(QueueConfig.PER_QUEUE_TTL_EXCHANGE_NAME, QueueConfig.DELAY_QUEUE_PER_QUEUE_TTL_NAME, null,
"触发异常，延迟再来".getBytes());
}
}
}
重点总结
- 创建mq配置的时候一定要注意绑定DLX和消费队列，如果需要用到异常处理的缓冲队列也一定要记得绑定对应的关系
- 个人认为推送消息时的messge非常重要，可以放想要使用的参数上去，然后消费的时候可以使用参数做处理，会方便很多
- 其实精华就是两个队列是一组，一个队列用来缓存消息并绑定超时时间，另一个队列是真正的消费队列，中间用一个DLX关联起来，缓存队列绝对不能被直接消费
此致，敬礼！！！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/1467.html
