这篇文章给大家介绍RabbitMQ中怎么实现延迟队列,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。
3.6在RabbitMQ。x之前我们一般采用死信队列+ TTL过期时间来实现延迟队列,我们这里不做过多介绍,可以参考之前文章来了解:TTL,死信队列
3.6在RabbitMQ。x开始,RabbitMQ官方提供了延迟队列的插件,可以下载放置到RabbitMQ根目录下的插件下。延迟队列插件下载
首先我们创建交换机和消息队列
进口org.springframework.amqp.core。*;进口org.springframework.context.annotation.Bean;进口org.springframework.context.annotation.Configuration;进口java.util.HashMap;进口java.util.Map; @Configurationpublic类MQConfig{,公共静态最终字符串LAZY_EXCHANGE=癊x.LazyExchange";最后,公共静态字符串LAZY_QUEUE=癕Q.LazyQueue";最后,公共静态字符串LAZY_KEY=袄痢?“;@ bean大敌;公共TopicExchange lazyExchange () {,,//Map<字符串,Object>优点=new HashMap<在();,,//设置交换机支持延迟消息推送,,//pros.put (“x-delayed-message",“topic");,,TopicExchange交换=new TopicExchange (LAZY_EXCHANGE,真的,假的,优点);,,exchange.setDelayed(真正的);,,返回交换;@ bean,,}公共队列lazyQueue () {,,返回新的队列(LAZY_QUEUE,真实);@ bean,,}公共绑定lazyBinding () {,,返回BindingBuilder.bind (lazyQueue ()), (lazyExchange ()); (LAZY_KEY);,}}
我们在交换的声明中可以设置exchange.setDelayed(真正的)来开启延迟队列,也可以设置为以下内容传入交换机声明的方法中,因为第一种方式的底层就是通过这种方式来实现的。
//Map<字符串,Object>优点=new HashMap<在();,,//设置交换机支持延迟消息推送,,//pros.put (“x-delayed-message",“topic");,,TopicExchange交换=new TopicExchange (LAZY_EXCHANGE,真的,假的,优点);
发送消息时我们需要指定延迟推送的时间,我们这里在发送消息的方法中传入参数新MessagePostProcessor()是为了获得消息对象,因为需要借助消息对象的api来设置延迟时间。
进口com.anqi.mq.config.MQConfig;进口org.springframework.amqp.AmqpException;进口org.springframework.amqp.core.Message;进口org.springframework.amqp.core.MessageDeliveryMode;进口org.springframework.amqp.core.MessagePostProcessor;进口org.springframework.amqp.rabbit.connection.CorrelationData;进口org.springframework.amqp.rabbit.core.RabbitTemplate;进口org.springframework.beans.factory.annotation.Autowired;进口org.springframework.stereotype.Component;进口java.util.Date; @Componentpublic类MQSender {@ autowired大敌;私人RabbitTemplate RabbitTemplate;,//confirmCallback returnCallback代码省略,请参照上一篇,公共空间sendLazy(对象消息){,,rabbitTemplate.setMandatory(真正的);,,rabbitTemplate.setConfirmCallback (confirmCallback);,,rabbitTemplate.setReturnCallback (returnCallback);,,//id +时间戳全局唯一,,CorrelationData CorrelationData=https://www.yisu.com/zixun/new CorrelationData(“12345678909”+新日期());//发送消息时指定头延迟时间rabbitTemplate.convertAndSend (MQConfig。懒惰LAZY_EXCHANGE。”引导”,消息,新MessagePostProcessor () {@Override postProcessMessage公共消息(消息消息)抛出AmqpException{//设置消息持久化message.getMessageProperties () .setDeliveryMode (MessageDeliveryMode.PERSISTENT);//message.getMessageProperties ()。setHeader (“x-delay”、“6000”);message.getMessageProperties () .setDelay (6000);返回消息;}},correlationData);}}
我们可以观察setDelay(整数)底层代码,也是在标题中设置x-delay。等同于我们手动设置标题
message.getMessageProperties () .setHeader (“x-delay",“6000“);
/* * *设置x-delay头。* @param延迟延误。* @since 1.6 */公共空setDelay(整数延迟){,如果延迟==null | |延迟& lt;0){,,this.headers.remove (X_DELAY);其他,},{,this.headers。把(X_DELAY,延迟);,}}
消费端进行消费
进口com.rabbitmq.client.Channel;进口org.springframework.amqp.rabbit.annotation。*;进口org.springframework.amqp.support.AmqpHeaders;进口org.springframework.stereotype.Component;进口java.io.IOException;进口java.util.Map; @Componentpublic类MQReceiver {, @RabbitListener(队列=癕Q.LazyQueue"), @RabbitHandler,公共空间>关于RabbitMQ中怎么实现延迟队列就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看的到。
