SpringBoot RabbitMQ 延迟队列代码实现

场景

用户下单后,如果30min未支付,则删除该订单,这时候就要可以用延迟队列

准备

利用rabbitmq_delayed_message_exchange插件;

首先下载该插件:https://www.rabbitmq.com/community-plugins.html

然后把该插件放到rabbitmq安装目录plugins下;

进入到sbin目录下,执行"rabbitmq-plugins.bat enable rabbitmq_delayed_message_exchange";

关闭RabbitMQ服务,然后再启动(直接重启该插件可能会不生效)。

SpringBoot RabbitMQ代码

application.properties配置文件

spring.application.name=spring-boot-rabbitmq
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=nut
spring.rabbitmq.password=nut

配置类

注意这里的"x-delayed-type"和"x-delayed-message"

/**
 * 延迟队列配置exchange
 */
@Configuration
public class DelayQueueConfig {

    public static final String DELAY_EXCHANGE = "DELAY_EXCHANGE";
    public static final String DELAY_QUEUE = "DELAY_QUEUE";
    public static final String DELAY_ROUTING_KEY = "DELAY_ROUTING_KEY";

    @Bean("delayExchange")
    public Exchange delayExchange() {
        Map<String, Object> args = new HashMap<>(1);
//       x-delayed-type    声明 延迟队列Exchange的类型
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE, "x-delayed-message",true, false,args);
    }

    @Bean("delayQueue")
    public Queue delayQueue() {
        return QueueBuilder.durable(DELAY_QUEUE).build();
    }

    /**
     * 将延迟队列通过routingKey绑定到延迟交换器
     *
     * @return
     */
    @Bean
    public Binding delayQueueBindExchange() {
        return new Binding(DELAY_QUEUE, Binding.DestinationType.QUEUE, DELAY_EXCHANGE, DELAY_ROUTING_KEY, null);
    }

}

生产者

发送消息时,指定延迟的毫秒

/**
 * 延迟队列发送者
 */
@Component
@Slf4j
public class DelayQueueSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendDelayQueue(int number) {
        log.warn("延迟队列发送 : {} milliseconds", number);
        // 这里的Exchange可以是业务的Exchange,为了方便测试这里直接往死信Exchange里投递消息
        rabbitTemplate.convertAndSend(
                DelayQueueConfig.DELAY_EXCHANGE,
                DelayQueueConfig.DELAY_ROUTING_KEY,
                number, (message) -> {
                    // 设置延迟的毫秒数
                    message.getMessageProperties().setDelay(number);
                    log.info("Now : {}", ZonedDateTime.now());
                    return message;
                });
    }
}

消费者

/**
 * 延迟队列消费者
 */
@Component
@Slf4j
@RabbitListener(queues = DelayQueueConfig.DELAY_QUEUE)
public class DelayQueueConsumer {

    @RabbitHandler
    public void receiveDelayMessage(Integer milliseconds){
        log.warn("DelayQueueConsumer Time : {}, and the millis : {}", ZonedDateTime.now(), milliseconds);

    }

}

测试

先启动项目;

然后在测试类中发送消息;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {

    @Autowired
    private DelayQueueSender delayQueueSender;

    @Test
    public void testDelayQueueSender(){
        delayQueueSender.sendDelayQueue(5000);
    }
}

发送消息窗口:

消费者受到消息:

时间间隔证明延迟队列发送完成!

参考:

https://blog.csdn.net/linsongbin1/article/details/80178122

https://blog.csdn.net/youjin/article/details/82586888

https://docs.spring.io/spring-amqp/docs/2.0.0.M2/reference/htmlsingle/#delayed-message-exchange

https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/

原文地址:https://www.cnblogs.com/theRhyme/p/10986409.html

时间: 2024-11-10 12:47:37

SpringBoot RabbitMQ 延迟队列代码实现的相关文章

C# RabbitMQ延迟队列功能实战项目演练

一.需求背景 当用户在商城上进行下单支付,我们假设如果8小时没有进行支付,那么就后台自动对该笔交易的状态修改为订单关闭取消,同时给用户发送一份邮件提醒.那么我们应用程序如何实现这样的需求场景呢?在之前的<C# Redis缓存过期实现延迟通知实战演练>分享课程中阿笨最后总结的时候说过Redis Pub/Sub是一种并不可靠地消息机制,他不会做信息的存储,只是在线转发,那么肯定也没有ack确认机制,另外只有订阅段监听时才会转发!我们是否有更好的方式去实现呢?今天给大家分享的比较好的解决方案就是通过

SpringBoot:RabbitMQ 延迟队列

SpringBoot 是为了简化 Spring 应用的创建.运行.调试.部署等一系列问题而诞生的产物,自动装配的特性让我们可以更好的关注业务本身而不是外部的XML配置,我们只需遵循规范,引入相关的依赖就可以轻易的搭建出一个 WEB 工程 初探RabbitMQ消息队列中介绍了RabbitMQ的简单用法,顺带提及了下延迟队列的作用.所谓延时消息就是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费. 延迟队列 延迟队列能做什么? 订单业务: 在电商/点餐

C#实现rabbitmq 延迟队列功能

最近在研究rabbitmq,项目中有这样一个场景:在用户要支付订单的时候,如果超过30分钟未支付,会把订单关掉.当然我们可以做一个定时任务,每个一段时间来扫描未支付的订单,如果该订单超过支付时间就关闭,但是在数据量小的时候并没有什么大的问题,但是数据量一大轮训数据库的方式就会变得特别耗资源.当面对千万级.上亿级数据量时,本身写入的IO就比较高,导致长时间查询或者根本就查不出来,更别说分库分表以后了.除此之外,还有优先级队列,基于优先级队列的JDK延迟队列,时间轮等方式.但如果系统的架构中本身就有

实现rabbitmq 延迟队列功能

最近在研究rabbitmq,项目中有这样一个场景:在用户要支付订单的时候,如果超过30分钟未支付,会把订单关掉.当然我们可以做一个定时任务,每个一段时间来扫描未支付的订单,如果该订单超过支付时间就关闭,但是在数据量小的时候并没有什么大的问题,但是数据量一大轮训数据库的方式就会变得特别耗资源.当面对千万级.上亿级数据量时,本身写入的IO就比较高,导致长时间查询或者根本就查不出来,更别说分库分表以后了.除此之外,还有优先级队列,基于优先级队列的JDK延迟队列,时间轮等方式.但如果系统的架构中本身就有

RabbitMQ 延迟队列实现订单支付结果异步阶梯性通知

在第三方支付中,例如支付宝.或者微信,对于订单请求,第三方支付系统采用的是消息同步返回.异步通知+主动补偿查询的补偿机制. 由于互联网通信的不可靠性,例如双方网络.服务器.应用等因素的影响,不管是同步返回.异步通知.主动查询报文都可能出现超时无响应.报文丢失等情况,所以像支付业务,对结果的通知一般采用几种方案结合的补偿机制,不能完全依赖某一种机制. 例如一个支付结果的通知,一方面会在支付页面跳转时候返回支付结果(一般只用作前端展示使用,非最终状态),同时会采用后台异步通知机制(有前台.后台通知的

Spring Boot(十四)RabbitMQ延迟队列

一.前言 延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单:2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度:3.过1分钟给新注册会员的用户,发送注册邮件等. 实现延迟队列的方式有两种: 通过消息过期后进入死信交换器,再由交换器转发到延迟消费队列,实现延迟功能: 使用rabbitmq-delayed-message-exchange插件实现延迟功能: 注意: 延迟插件rabbitmq-delayed-message-exchange是在RabbitMQ 3.5.7及以上

rabbitmq延迟队列demo

工程结构: 定义jar包依赖的版本,版本很重要,rabbit依赖spring,必须一致,否则报错: <properties> <springframework.version>4.2.7.RELEASE</springframework.version> <spring-rabbit.version>1.6.1.RELEASE</spring-rabbit.version> <junit.version>4.12</junit.

【RabbitMQ】一文带你搞定RabbitMQ延迟队列

本文口味:鱼香肉丝? ?预计阅读:10分钟 一.说明 在上一篇中,介绍了RabbitMQ中的死信队列是什么,何时使用以及如何使用RabbitMQ的死信队列.相信通过上一篇的学习,对于死信队列已经有了更多的了解,这一篇的内容也跟死信队列息息相关,如果你还不了解死信队列,那么建议你先进行上一篇文章的阅读. 这一篇里,我们将继续介绍RabbitMQ的高级特性,通过本篇的学习,你将收获: 什么是延时队列 延时队列使用场景 RabbitMQ中的TTL 如何利用RabbitMQ来实现延时队列 二.本文大纲

RabbitMQ延迟队列简单示例

简介 延迟队列存储的消息是不希望被消费者立刻拿到的,而是等待特定时间后,消费者才能拿到这个消息进行消费.使用场景比较多,例如订单限时30分钟内支付,否则取消,再如分布式环境中每隔一段时间重复执行某操作. 下面举一个简单的例子,例子大概意思是分别在首次发送消息后的10秒.40秒.100秒后重新读取到消息.为了直观,不使用RabbitMQ其他多余的特性. 准备工作 在Centos7下安装RabbitMQ,版本为3.6.12单机版本(非集群),IP是127.0.0.1,端口是15672,使用web管理