网站中的幻灯片ie6显示 ie7如何兼容,米课wordpress建站,兰州的互联网公司有哪些,丹东黄页网RabbitMQ延迟队列的实现 延迟消息是什么延迟消息的实现死信交换机代码实现 延迟消息插件 延迟消息是什么
延迟消息是将消息发送到MQ中#xff0c;消费者不会立即收到消息#xff0c;而是过一段时间之后才会收到消息#xff0c;进行处理。在一些业务中#xff0c;可以用到延… RabbitMQ延迟队列的实现 延迟消息是什么延迟消息的实现死信交换机代码实现 延迟消息插件 延迟消息是什么
延迟消息是将消息发送到MQ中消费者不会立即收到消息而是过一段时间之后才会收到消息进行处理。在一些业务中可以用到延迟消息比如我们在成功下单一个商品后需要立即付款为了避免商品库存一直被占有我们会给商品设置一个支付时间如果在这段时间没有支付成功就会恢复库存删除订单对于订单支付的超时删除我们是通过延迟消息来实现的让消费者在支付超时之后查询用户是否支付如果支付成功直接返回如果支付失败就恢复库存删除订单。
延迟消息的实现
延迟消息由以下两种方式实现第一种是通过绑定死信交换机实现第二种通过延迟消息插件实现推荐使用第二种更加简单
死信交换机
满足以下三种情况之一的叫做死信 1、在设置了过期时间的消息放入队列中超过了过期时间没有被处理的消息 2、消息消费失败返回nack或者reject并且不能重复入队 3、队列消息堆积满了最早的消息叫做死信 我们可以给队列绑定参数指定交换机那么死信会被投递到指定交换机。 消息队列实现原理我们可以设置一组没有消费者的交换机和队列设置另一组处理绑定死信的交换机、队列和消费者来处理延迟消息。
代码实现
定义死信交换机等和消费延迟消息交换机等 Configuration
public class DelayConfiguration {/*** 定义死信交换机、队列以及绑定*/Beanpublic DirectExchange exchange() {return new DirectExchange(dead.direct);}Beanpublic Queue queue() {Queue queue new Queue(dead.queue);queue.addArgument(x-dead-letter-exchange, delay.direct);return queue;}Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(exchange()).with(dead);}/*** 定义处理延迟消息的交换机、队列和绑定*/Beanpublic DirectExchange exchange1() {return new DirectExchange(delay.direct);}Beanpublic Queue queue1() {return new Queue(delay.queue);}Beanpublic Binding binding1() {return BindingBuilder.bind(queue1()).to(exchange1()).with(dead);}
}
定义延迟消息监听器 RabbitListener(queues delay.queue)public void listen(String msg){log.info(LocalDateTime.now(): msg);}测试
Testvoid sendDeadMsg() {rabbitTemplate.convertAndSend(dead.direct, dead, 我是死信, new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {
// 设置过期消息时间message.getMessageProperties().setExpiration(10000);return message;}});}结果 消费者在十秒钟后成功消费延迟消息
延迟消息插件
我们在之前通过死信交换机来实现延迟队列但是死信交换机是专门用来存放无法处理的消息并且使用死信交换机实现过于复杂我们需要手动定义两个交换机和队列因而RabbitMQ提供了延迟消息插件来让我们更简单的实现延迟消息。 原理给消息设置延迟时间当将消息放入MQ时MQ的交换机不会立即将消息放入队列而是会在交换机中暂存延迟时间过后将消息路由到队列中可以让队列处理延迟消息。 安装插件插件安装可以借鉴这篇博客 代码实现 消费者 RabbitListener(bindings QueueBinding(value Queue(value delay.queue,durable true),// 开启延迟交换机exchange Exchange(name delay.direct,delayed true),key dead))public void listen(String msg){log.info(msg);}Testvoid sendDeadMsg() {rabbitTemplate.convertAndSend(delay.direct, delay, 我是死信, new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {
// 设置延迟消息时间message.getMessageProperties().setDelay(10000);return message;}});}
