RabbitMQ延迟队列

zoeil 2023-04-18 原文

一、概念

延时队列， 队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

二、使用场景

订单在十分钟之内未支付则自动取消
新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
用户注册成功后，如果三天内没有登陆则进行短信提醒。
用户发起退款，如果三天内没有得到处理则通知相关运营人员。
预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：

发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；看起来似乎使用 定时任务 ，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？ 如果数据量比较少，确实可以这样做 ，比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。

但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。

三、RabbitMQ 中的 TTL

TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有 消息的最大存活时间， 单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为 " 死信 " 。如果同时配置了队列的 TTL 和消息的TTL，那么较小的那个值将会被使用，有两种方式设置 TTL 。

（一）队列设置 TTL

第一种是在创建队列的时候设置队列的“x-message-ttl”属性

Map<String, Object> arguments = new HashMap<>();
// 声明队列的TTL
arguments.put("x-message-ttl", 10000);
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();

（二）消息设置 TTL

另一种方式便是针对每条消息设置 TTL

rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
    msg.getMessageProperties().setExpiration(ttl);
    return msg;
});

（三）两者的区别

如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃 ( 如果配置了死信队列被丢到死信队列中) ，而第二种方式，消息即使过期，也不一定会被马上丢弃，因为 消息是否过期是在即将投递到消费者 之前判定的 ， 如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间 ；另外，还需要注意的一点是，如果不设置 TTL ，表示消息永远不会过期， 如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

前一小节我们介绍了死信队列，刚刚又介绍了 TTL ，至此利用 RabbitMQ 实现延时队列的两大要素已经集齐，接下来只需要将它们进行融合，再加入一点点调味料，延时队列就可以新鲜出炉了。想想看，延时队列，不就是想要消息延迟多久被处理吗，TTL 则刚好能让消息在延迟多久之后成为死信，另一方面，成为死信的消息都会被投递到死信队列里，这样只需要消费者一直消费死信队列里的消息就完事了，因为里面的消息都是希望被立即处理的消息。

四、整合SpringBoot实现延迟队列

（一）创建项目

（二）添加依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!--RabbitMQ 依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--swagger-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>
        <!--RabbitMQ 测试依赖-->
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>

（三）修改配置文件

spring.rabbitmq.host=192.168.23.100
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

（四）添加Swagger配置类

@Configuration
@EnableSwagger2
public class SwaggerConfig {

    @Bean
    public Docket webApiConfig() {
        return new Docket(DocumentationType.SWAGGER_2)
                .groupName("webapi")
                .apiInfo(webApiInfo())
                .select()
                .build();
    }

    public ApiInfo webApiInfo() {
        return new ApiInfoBuilder()
                .title("rabbitmq 接口文档")
                .description("本文档描述了 rabbitmq 微服务接口定义")
                .version("1.0")
                .contact(new Contact("enjoy6288", "http://atguigu.com",
                        "1551388580@qq.com"))
                .build();
    }
}

五、队列TTL

（一）代码架构图

创建两个队列 QA 和 QB ，两者队列 TTL 分别设置为 10S 和 40S ，然后在创建一个交换机 X 和死信交换机 Y ，它们的类型都是 direct ，创建一个死信队列 QD ，它们的绑定关系如下：

（二）配置文件类

@Configuration
public class TtlQueueConfig {
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String DEAD_LETTER_QUEUE = "QD";

    // 声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }
    // 声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    // 声明队列A
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> arguments = new HashMap<>();
        // 当前队列的死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        // 当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "YD");
        // 声明队列的TTL
        arguments.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    }

    // 声明队列A绑定交换机X
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange")DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }


    // 声明队列B
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        // 当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "YD");
        // 声明队列的TTL
        arguments.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    }

    // 声明队列B绑定交换机X
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange")DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }



    // 声明死信队列
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    @Bean
    // 声明死信队列 QD 绑定关系
    public Binding queuedBindingY(@Qualifier("queueD")Queue queueD,
                                  @Qualifier("yExchange")DirectExchange exchange) {
        return BindingBuilder.bind(queueD).to(exchange).with("YD");
    }

}

（三）消息生产者

    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message) {
        log.info("当前时间是{}，发送一条信息给两个 TTL 队列:{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10s的队列" + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40s的队列" + message);
    }

（四）消息消费者

@Component
@Slf4j
public class DeadLetterQueueConsumer {

    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到死信队列信息{}", new Date().toString(), msg);
    }
}

发起一个请求 http://localhost:8080/ttl/sendMsg/嘻嘻嘻

第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 40S 之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了。

不过，如果这样使用的话，岂不是 每增加一个新的时间需求，就要新增一个队列 ，这里只有 10S 和 40S 两个时间选项，如果需要一个小时后处理，那么就需要增加 TTL 为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？

六、延迟队列优化

（一）代码架构图

在这里新增了一个队列 QC,绑定关系如下,该队列不设置 TTL 时间，而是由生产者设置过期时间

（二）配置文件类

@Configuration
public class TtlQueueConfig {
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String DEAD_LETTER_QUEUE = "QD";
    public static final String QUEUE_C = "QC";

    // 声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }
    // 声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    // 声明队列A
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> arguments = new HashMap<>();
        // 当前队列的死信交换机
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        // 当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "YD");
        // 声明队列的TTL
        arguments.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    }

    // 声明队列A绑定交换机X
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange")DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }


    // 声明队列B
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        // 当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "YD");
        // 声明队列的TTL
        arguments.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    }

    // 声明队列B绑定交换机X
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange")DirectExchange xExchange) {
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }

    // 声明队列C
    @Bean("queueC")
    public Queue queueC() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        // 当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "YD");
        return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
    }

    // 声明队列C绑定交换机X
    @Bean
    public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange")DirectExchange xExchange) {
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }



    // 声明死信队列
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    @Bean
    // 声明死信队列 QD 绑定关系
    public Binding queuedBindingY(@Qualifier("queueD")Queue queueD,
                                  @Qualifier("yExchange")DirectExchange exchange) {
        return BindingBuilder.bind(queueD).to(exchange).with("YD");
    }

}

（三）消息生产者

    @GetMapping("/sendExpirationMsg/{message}/{ttl}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttl) {
        log.info("当前时间是{}，发送一条过期信息给两个 TTL 队列:{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
            msg.getMessageProperties().setExpiration(ttl);
            return msg;
        });
    }

发起请求

http://localhost:8080/ttl/sendExpirationMsg/ 你好 1/20000

http://localhost:8080/ttl/sendExpirationMsg/ 你好 2/2000

看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期 ，如果过期则丢到死信队列， 如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行 。

七、Rabbitmq 插件实现延迟队列

关于插件的安装可以查看这篇文章Docker安装RabbitMq延迟队列插件

（一）代码架构图

在这里新增了一个队列 delayed.queue,一个自定义交换机 delayed.exchange，绑定关系如下:

（二）配置文件类

在我们自定义的交换机中，这是一种新的交换类型，该类型消息支持延迟投递机制消息传递后并

不会立即投递到目标队列中，而是存储在 mnesia( 一个分布式数据系统 ) 表中，当达到投递时间时，才投递到目标队列中。

（三）消息生产者

/*
 * 基于插件的延迟队列和延迟交换机
 */
@Configuration
public class DelayedQueueConfig {

    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";


    // 声明队列
    @Bean
    public Queue delayedQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    // 声明自定义交换机
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");

        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    // 声明队列和延迟交换机的绑定
    @Bean
    public Binding bindingDelayedQueue(@Qualifier("delayedQueue")Queue delayedQueue,
                                       @Qualifier("delayedExchange")CustomExchange exchange) {
        return BindingBuilder.bind(delayedQueue).to(exchange).with(DELAYED_ROUTING_KEY).noargs();
    }
}

（四）消息消费者

@Component
@Slf4j
public class DelayedQueueConsumer {

    @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
    public void receiveDelayedQueue(String message) {
        log.info("当前时间：{}， 接收到消息: {}", new Date().toString(), message);
    }
}

发起请求：

http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000

http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000

第二个消息被先消费掉了，符合预期

八、总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用

RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz或者利用 kafka 的时间轮，这些方式各有特点 , 看需要适用的场景。

RabbitMQ 延迟 span style 000000 分布式 java-rabbitmq 后端 java

有关RabbitMQ延迟队列的更多相关文章

ruby - 分布式事务和队列，ruby，erlang，scala - 2
我有一个涉及多台机器、消息队列和事务的问题。因此，例如用户点击网页，点击将消息发送到另一台机器，该机器将付款添加到用户的帐户。每秒可能有数千次点击。事务的所有方面都应该是容错的。我以前从未遇到过这样的事情，但一些阅读表明这是一个众所周知的问题。所以我的问题。我假设安全的方法是使用两阶段提交，但协议(protocol)是阻塞的，所以我不会获得所需的性能，我是否正确？我通常写Ruby，但似乎Redis之类的数据库和Rescue、RabbitMQ等消息队列系统对我的帮助不大——即使我实现某种两阶段提交，如果Redis崩溃，数据也会丢失，因为它本质上只是内存。所有这些让我开始关注erlang和
ruby-on-rails - 在所有延迟的作业之前 Hook - 2
是否可以在所有delayed_job任务之前运行一个方法？基本上，我们试图确保每个运行delayed_job的服务器都有我们代码的最新实例，所以我们想运行一个方法来在每个作业运行之前检查它。(我们已经有了“check”方法并在别处使用它。问题只是关于如何从delayed_job中调用它。) 最佳答案现在有一种官方方法可以通过插件来做到这一点。这篇博文通过示例清楚地描述了如何执行此操作http://www.salsify.com/blog/delayed-jobs-callbacks-and-hooks-in-rails(本文中描述
ruby-on-rails - Ruby 长时间运行的进程对队列事件使用react - 2
我有一个将某些事件写入队列的Rails3应用。现在我想在服务器上创建一个服务，每x秒轮询一次队列，并按计划执行其他任务。除了创建ruby脚本并通过cron作业运行它之外，还有其他稳定的替代方案吗？最佳答案尽管启动基于Rails的持久任务是一种选择，但您可能希望查看更有序的系统，例如delayed_job或Starling管理您的工作量。我建议不要在cron中运行某些东西，因为启动整个Rails堆栈的开销可能很大。每隔几秒运行一次它是不切实际的，因为Rails上的启动时间通常为5-15秒，具体取决于您的硬件。不过，每天这样做几
ruby - Heroku - 如何开始工作人员(延迟工作)？ - 2
我有一些使用delayed_job的小程序。在我的本地主机上一切正常，但是当我将我的应用程序部署到Heroku并单击应该由delayed_job执行的链接时，没有任何反应，“任务”只是保存到表delayed_job中。Inthisarticleonherokublog写入时，执行delayed_job表中的任务，当运行此命令时rakejobs:work。但是我怎样才能运行这个命令呢？命令应该放在哪里？在代码中，还是从终端控制台？最佳答案如果您正在运行Cedar堆栈，请从终端控制台运行以下命令:herokurunrakejobs:
ruby - 在不提供其所有属性的情况下获取队列 - 2
我正在尝试为现有队列编写消费者。RabbbitMQ在一个单独的实例中运行，名为“org-queue”的队列已经创建并绑定(bind)到一个交换器。org-queue是一个持久队列，它还有一些额外的属性。现在我需要从这个队列接收消息。我使用下面的代码来获取队列的实例conn=Bunny.newconn.startch=conn.create_channelq=ch.queue("org-queue")它抛出一个错误，指出不同的耐用属性。默认情况下，Bunny似乎使用durable=false。所以我添加了durabletrue作为参数。现在它说明了其他参数之间的区别。我是否需要指定所有参
ruby - 如何在特定队列中推送作业并使用 sidekiq 限制工作人员数量？ - 2
我知道我们可以做到:sidekiq_optionsqueue:"Foo"但在这种情况下，Worker只分配给一个队列:“Foo”。我需要在特定队列中分配作业(而不是worker)。使用Resque很容易:Resque.enqueue_to(queue_name,my_job)另外，为了并发问题，我需要限制每个队列的Worker数量为1。我该怎么做？最佳答案您可能会使用https://github.com/brainopia/sidekiq-limit_fetch然后:Sidekiq::Client.push({'class'=>
Python：每日一题之小张的衣服（优先队列、哈夫曼编码） - 2
题目描述小张买了 n 件白色的衣服，他觉得所有衣服都是一种颜色太单调，希望对这些衣服进行染色，每次染色时，他会将某种颜色的所有衣服寄去染色厂，第 i 件衣服的邮费为 ai 元，染色厂会按照小张的要求将其中一部分衣服染成同一种任意的颜色，之后将衣服寄给小张，请问小张要将 n 件衣服染成不同颜色的最小代价是多少？输入描述第一行为一个整数 n ，表示衣服的数量。第二行包括 n 个整数a1,a2...an 表示第 i 件衣服的邮费为 ai 元。（1≤n≤10^5,1≤ai≤10^9 ）输出描述输出一个整数表示小张所要花费的最小代价。输入输出样例输入551321输出25 思考🤔：题意：意思是
ruby-on-rails - 成功发送延迟作业电子邮件后更新用户表 - 2
我是一个尝试使用delayed_job的NOOB。我想在使用延迟作业成功发送邮件后更新用户模型。发送邮件:UserMailer.delay.welcome_email(user)如果邮件发送成功，请执行以下操作:User.update_attributes(:emailed=>true)邮件发送成功后如何回调或触发？最佳答案您需要创建一个Job对象而不是调用#delay帮助程序。您可以使用successHook来执行回调。classWelcomeEmailJob 关于ruby-on-
ruby-on-rails - 启动同一作业的多个延迟作业进程 - 2
我在运行多个工作器的设置中使用延迟作业。就我的问题而言，这并不重要，但假设我有10个worker(目前在开发模式下这样做)。我遇到的问题是两个不同的工作人员有时会开始处理同一个工作，调用我的工作对象的perform方法。据我所知，DelayedJob正在使用悲观锁定来防止这种情况发生，但有时它似乎仍然有足够的时间在第一个worker有时间实际锁定它之前锁定它。我只是想看看有没有其他人遇到过这个问题，或者是我的设置有问题。我正在使用Postrgres，这发生在我的开发机器和我托管它的Heroku上。我会尝试在我的工作中解决这个问题，但发生这种情况仍然有点问题。理想情况下，延迟作业永远不会
ruby - 延迟作业每次引发错误时都会创建空气制动器 - 2
defperformrefund_log={success:refund_retry.success?,amount:refund_amount,action:"refund"}ifrefund_retry.success?refund_log[:reference]=refund_retry.transaction.idrefund_log[:message]=refund_retry.transaction.statuselserefund_log[:message]=refund_retry.messagerefund_log[:params]={}refund_retry.er

RabbitMQ延迟队列

一、概念

二、使用场景

三、RabbitMQ 中的 TTL

（一）队列设置 TTL

（二）消息设置 TTL

（三）两者的区别

四、整合SpringBoot实现延迟队列

（一）创建项目

（二）添加依赖

（三）修改配置文件

（四）添加Swagger配置类

五、队列TTL

（一）代码架构图

（二）配置文件类

（三）消息生产者

（四）消息消费者

六、延迟队列优化

（一）代码架构图

（二）配置文件类

（三）消息生产者

七、Rabbitmq 插件实现延迟队列

（一）代码架构图

（二）配置文件类

（三）消息生产者

（四）消息消费者

八、总结

有关RabbitMQ延迟队列的更多相关文章

随机推荐