查询网站备案进度查询,山东省优质校建设网站,微盟登录,网页设计培训班机构文章目录 前言延迟队列延迟队列的概念TTL 死信队列模拟延迟队列设置队列的 TTL设置消息的 TTL 延迟队列插件安装并且启动插件服务使用插件实现延迟功能 前言
前面我们学习了 TTL 和死信队列#xff0c;当队列中的消息达到了过期时间之后#xff0c;那么这个消息就会被死信交… 文章目录 前言延迟队列延迟队列的概念TTL 死信队列模拟延迟队列设置队列的 TTL设置消息的 TTL 延迟队列插件安装并且启动插件服务使用插件实现延迟功能 前言
前面我们学习了 TTL 和死信队列当队列中的消息达到了过期时间之后那么这个消息就会被死信交换机投递到死信队列中然后由专门的消费者进行消费这篇文章将为大家介绍 RabbitMQ 的另一高级特性——延迟队列。
延迟队列
延迟队列的概念
延迟队列通常指的是一种队列其中的消息在发送后不会立即被消费而是会等待一段时间即延迟时间后才变得可消费。
RabbitMQ 本身不直接提供名为“延迟队列”的内置功能但我们可以利用 RabbitMQ 的一些特性和机制来模拟或实现延迟队列的行为。就比如 TTL 死信队列方式模拟出延迟队列的功能。
TTL 死信队列模拟延迟队列 我们的消费者不去消费正常队列中的消息而是去消费死信队列中的消息为什么这样能实现延迟对立的功能呢当生产者生产消息投递给正常交换机之后交换机会将这个消息根据 routing key 和 binding key 路由到指定的队列中因为这个正常队列没有消费者可以消费消息所以到达了了这个队列中的消息就会存储在这个队列中因为这个队列设置了 TTL所以消息到达了过期时间之后就会被投递给死信交换机 DLX然后死信交换机就会将这个消息投递给 DLQ然后消费者从这个 DLQ 中去消费消息这样消费者就会在生产者生产的消息到达队列之后不会立即去消费而是会等待一段时间这样就通过 TTL 死信队列的方式模拟出了延迟队列的功能。
设置队列的 TTL
那么我们看看通过代码如何实现
public class Constants {public static final String NORMAL_EXCHANGE normal.exchange;public static final String NORMAL_QUEUE normal.queue;public static final String DL_EXCHANGE dl.exchange;public static final String DL_QUEUE dl.queue;
}声明交换机、队列和绑定关系
Configuration
public class DLConfig {Bean(normalExchange)public Exchange normalExchange() {return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).build();}Bean(normalQueue)public Queue normalQueue() {return QueueBuilder.durable(Constants.NORMAL_QUEUE).ttl(1000).deadLetterExchange(Constants.DL_EXCHANGE).deadLetterRoutingKey(dl).build();}Bean(normalBinding)public Binding normalBinding(Qualifier(normalExchange) Exchange exchange, Qualifier(normalQueue) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(normal).noargs();}Bean(DLExchange)public Exchange DLExchange() {return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).build();}Bean(DLQueue)public Queue DLQueue() {return QueueBuilder.durable(Constants.DL_QUEUE).build();}Bean(DLBinding)public Binding DLBinding(Qualifier(DLExchange) Exchange exchange,Qualifier(DLQueue) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(dl).noargs();}
}生产者代码
RequestMapping(producer)
RestController
public class ProducerController {Autowiredprivate RabbitTemplate rabbitTemplate;RequestMapping(delay)public String delay() {rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,rabbitmq delay,发送时间: new Date());return 消息发送成功;}
}消费者
Component
public class DelayListener {RabbitListener(queues Constants.DL_QUEUE)public void listener(String message, Channel channel) {System.out.println(时间: new Date() ,接收到消息: message channel);}
}可以看到通过 TTL 死信队列的方式是可以实现延迟队列的功能的虽然可能从消息发送的时间到消费者消费到这个消息的时间可能会比预想的时间多一点中间消息传输花费了一点时间。
这是一次发送一条消息的情况如果我们一次发多条信息看看什么效果
RequestMapping(delay)
public String delay() throws InterruptedException {rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,rabbitmq delay1,发送时间: new Date());Thread.sleep(1000);rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,rabbitmq delay2,发送时间: new Date());Thread.sleep(1000);rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal,rabbitmq delay3,发送时间: new Date());return 消息发送成功;
}当为队列设置 TTL 的时候消息会按照进队列的先后顺序进行消费这是为队列设置 TTL 的情况为队列设置 TTL 意味着进入该队列中的所有消息都具有过期时间且过期时间如果没有单独设置消息的过期时间的话那么该队列中的消息的过期时间都是相同的那么如果我们想要每个消息的过期时间都不相同的话就需要单独给消息设置过期时间接下来我们看看给消息单独设置过期时间的情况。
设置消息的 TTL
public static final String NORMAL_QUEUE2 normal.queue2;创建出一个没有设置过期时间的普通队列
Bean(normalQueue2)
public Queue normalQueue2() {return QueueBuilder.durable(Constants.NORMAL_QUEUE2).deadLetterExchange(Constants.DL_EXCHANGE).deadLetterRoutingKey(dl).build();
}Bean(normalBinding2)
public Binding normalBinding2(Qualifier(normalExchange) Exchange exchange,Qualifier(normalQueue2) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(normal2).noargs();
}设置消息的过期时间
RequestMapping(delay2)
public String delay2(){rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal2,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setExpiration(10000);return message;});rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal2,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setExpiration(20000);return message;});return 消息发送成功;
}消费者的代码就不需要改动 为消息设置 TTL 就可以时间每个消息的过期时间都不相同但是这里会有一个问题就是如果先投递的消息的过期时间如果晚于后面投递的消息的过期时间就会出现问题
RequestMapping(delay2)
public String delay2(){rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal2,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setExpiration(20000);return message;});rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,normal2,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setExpiration(10000);return message;});return 消息发送成功;
}可以发现设置了过期时间为 10s 的消息也是过了 20s 之后才进入到死信队列然后才被消费。
消息过期之后不一定会被马上丢弃。因为RabbitMQ只会检查队首消息是否过期如果过期则丢到死信队列。 此时就会造成一个问题如果第一个消息的延时时间很长第二个消息的延时时间很短那第二个消息并不会优先得到执行。因为我们的普通队列并没有消费者消费其中的消息我们都知道对消息设置过期时间只有在消息被快要使用的之前才会判断它是否过期但是这里由于队列没有消费者所以消息也就不会被使用到那么 RabbitMQ 就只会检查队首的消息是否过期这样就导致了队首的消息没有过期那么之后的消息也不会被检查到过期投递到死信队列中。
所以在考虑使用TTL死信队列实现延迟任务队列的时候需要确认业务上每个任务的延迟时间是一致的如果遇到不同的任务类型需要不同的延迟的话需要为每一种不同延迟时间的消息建立单独的消息队列。
延迟队列插件
虽然 RabbitMQ 没有直接实现延迟队列但是 RabbitMQ 提供了一个延迟的插件来实现延迟的功能。
具体的可以看看官方文档 https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq#delaying-messages
安装并且启动插件服务 也可以点击这里 我们下载的插件的版本需要和 RabbitMQ 的版本兼容 版本对应关系 下载完成插件之后可以在这里查看如何配置https://www.rabbitmq.com/docs/installing-plugins 我们将下载完成的文件放在 RabbitMQ 的插件目录下
我们使用的是 Linux所以需要将插件放在 /usr/lib/rabbitmq/plugins 或者 /usr/lib/rabbitmq/lib/rabbitmq_server-version/plugins文件下如果没有这个路径可以自己创建 下载完成插件并且将其放在指定目录下之后我们就可以启动插件了
使用 rabbitmq-plugins list 查看 rabbitmq 的插件
如果我们之前启动了一个错误版本的插件可以使用 rabbitmq-plugins disbale 插件名称 来禁用插件 使用 rabbitmq-plugins enable 插件名称 来启动插件服务 如何验证插件已经安装并且启动成功我们在 rabbitmq 的管理页面新建一个交换机的时候看看交换机的类型是否有 x-delayed-message 这个类型
我们刷新一下管理页面然后在新建交换机的时候查看交换机的类型 这就说明我们的延迟插件启动成功了。
使用插件实现延迟功能
安装并且启动延迟插件之后我们来看看代码如何实现一个延迟功能的队列
public static final String DELAY_EXCHANGE delay.exchange;
public static final String DELAY_QUEUE delay.queue;声明交换机、对立和绑定关系
Configuration
public class DelayConfig {Bean(delayExchange)public Exchange delayExchange() {return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build();}Bean(delayQueue)public Queue delayQueue() {return QueueBuilder.durable(Constants.DELAY_QUEUE).build();}Bean(delayBinding)public Binding delayBinding(Qualifier(delayExchange) Exchange exchange, Qualifier(delayQueue) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(delay).noargs();}
}生产者代码
RequestMapping(delay3)
public String delay3(){rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,delay,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setDelay(20000);return message;});rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,delay,rabbitmq delay,发送时间 new Date(),message - {message.getMessageProperties().setDelay(10000);return message;});return 消息发送成功;
}消费者代码
RabbitListener(queues Constants.DELAY_QUEUE)
public void listener2(String message, Channel channel) {System.out.println(时间: new Date() ,接收到消息: message channel);
}启动项目之后会创建一个 x-delay-message 类型的交换机 使用延迟插件的话就可以使得队列中的消息可以按照延迟时间到达消费者。
