关于java:RabbitMQ-实现延迟队列的两种方式

47次阅读

共计 5564 个字符，预计需要花费 14 分钟才能阅读完成。

@[toc]
定时工作各种各样，常见的定时工作例如日志备份，咱们可能在每天凌晨 3 点去备份，这种固定工夫的定时工作咱们个别采纳 cron 表达式就能轻松的实现，还有一些比拟非凡的定时工作，向大家看电影中的定时炸弹，3 分钟后爆炸，这种定时工作就不太好用 cron 去形容，因为开始工夫不确定，咱们开发中有的时候也会遇到相似的需要，例如：

  • 在电商我的项目中，当咱们下单之后，个别须要 20 分钟之内或者 30 分钟之内付款，否则订单就会进入异样解决逻辑中，被勾销，那么进入到异样解决逻辑中，就能够当成是一个提早队列。
  • 我买了一个智能砂锅，能够用来煮粥，下班前把素材都放到锅里，而后设置几点几分开始煮粥，这样上班后就能够喝到香喷喷的粥了，那么这个煮粥的指令也能够看成是一个提早工作，放到一个提早队列中，工夫到了再执行。
  • 公司的会议预约零碎，在会议预约胜利后，会在会议开始前半小时告诉所有预约该会议的用户。
  • 平安工单超过 24 小时未解决，则主动拉企业微信群揭示相干责任人。
  • 用户下单外卖当前，间隔超时工夫还有 10 分钟时揭示外卖小哥行将超时。
  • …

很多场景下咱们都须要提早队列。

本文以 RabbitMQ 为例来和大家聊一聊提早队列的玩法。

整体上来说，在 RabbitMQ 上实现定时工作有两种形式：

  • 利用 RabbitMQ 自带的音讯过期和私信队列机制，实现定时工作。
  • 应用 RabbitMQ 的 rabbitmq_delayed_message_exchange 插件来实现定时工作，这种计划较简略。

两种用法咱们别离来看。

1. 用插件

1.1 装置插件

首先咱们须要下载 rabbitmq_delayed_message_exchange 插件，这是一个 GitHub 上的开源我的项目，咱们间接下载即可：

  • https://github.com/rabbitmq/r…

抉择适宜本人的版本，我这里抉择最新的 3.9.0 版。

下载实现后在命令行执行如下命令将下载文件拷贝到 Docker 容器中去：

docker cp ./rabbitmq_delayed_message_exchange-3.9.0.ez some-rabbit:/plugins

这里第一个参数是宿主机上的文件地址，第二个参数是拷贝到容器的地位。

接下来再执行如下命令进入到 RabbitMQ 容器中：

docker exec -it some-rabbit /bin/bash

进入到容器之后，执行如下命令启用插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

启用胜利之后，还能够通过如下命令查看所有装置的插件，看看是否有咱们刚刚装置过的插件，如下：

rabbitmq-plugins list

命令的残缺执行过程如下图：

OK，配置实现之后，接下来咱们执行 exit 命令退出 RabbitMQ 容器。而后开始编码。

1.2 音讯收发

接下来开始音讯收发。

首先咱们创立一个 Spring Boot 我的项目，引入 Web 和 RabbitMQ 依赖，如下：

我的项目创立胜利后，在 application.properties 中配置 RabbitMQ 的根本信息，如下：

spring.rabbitmq.host=localhost
spring.rabbitmq.password=guest
spring.rabbitmq.username=guest
spring.rabbitmq.virtual-host=/

接下来提供一个 RabbitMQ 的配置类：

@Configuration
public class RabbitConfig {
    public static final String QUEUE_NAME = "javaboy_delay_queue";
    public static final String EXCHANGE_NAME = "javaboy_delay_exchange";
    public static final String EXCHANGE_TYPE = "x-delayed-message";

    @Bean
    Queue queue() {return new Queue(QUEUE_NAME, true, false, false);
    }

    @Bean
    CustomExchange customExchange() {Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_NAME, EXCHANGE_TYPE, true, false,args);
    }
    
    @Bean
    Binding binding() {return BindingBuilder.bind(queue())
                .to(customExchange()).with(QUEUE_NAME).noargs();}
}

这里次要是交换机的定义有所不同，小伙伴们须要留神。

这里咱们应用的交换机是 CustomExchange，这是一个 Spring 中提供的交换机，创立 CustomExchange 时有五个参数，含意别离如下：

  • 交换机名称。
  • 交换机类型，这个中央是固定的。
  • 交换机是否长久化。
  • 如果没有队列绑定到交换机，交换机是否删除。
  • 其余参数。

最初一个 args 参数中，指定了交换机音讯散发的类型，这个类型就是大家熟知的 direct、fanout、topic 以及 header 几种，用了哪种类型，未来交换机散发音讯就按哪种形式来。

接下来咱们再创立一个音讯消费者：

@Component
public class MsgReceiver {private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
    public void handleMsg(String msg) {logger.info("handleMsg,{}",msg);
    }
}

打印一下音讯内容即可。

接下来再写一个单元测试办法来发送音讯：

@SpringBootTest
class MqDelayedMsgDemoApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() throws UnsupportedEncodingException {Message msg = MessageBuilder.withBody(("hello 江南一点雨"+new Date()).getBytes("UTF-8")).setHeader("x-delay", 3000).build();
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.QUEUE_NAME, msg);
    }

}

在音讯头中设置音讯的延迟时间。

好啦，接下来启动 Spring Boot 我的项目，而后运行单元测试办法发送音讯，最终的控制台打印日志如下：

从日志中能够看到音讯提早曾经实现了。

2. DLX 实现提早队列

2.1 提早队列实现思路

提早队列实现的思路也很简略，就是上篇文章咱们所说的 DLX（死信交换机）+TTL（音讯超时工夫）。

咱们能够把死信队列就当成提早队列。

具体来说是这样：

如果一条音讯须要提早 30 分钟执行，咱们就设置这条音讯的有效期为 30 分钟，同时为这条音讯配置死信交换机和死信 routing_key，并且不为这个音讯队列设置消费者，那么 30 分钟后，这条音讯因为没有被消费者生产而进入死信队列，此时咱们有一个消费者就在“蹲点”这个死信队列，音讯一进入死信队列，就立马被生产了。

这就是提早队列的实现思路，是不是很简略？

2.2 案例

接下来松哥通过一个简略的案例，来和大家演示一下提早队列的具体实现。

首先筹备好一个启动的 RabbitMQ。

而后咱们创立一个 Spring Boot 我的项目，引入 RabbitMQ 依赖：

而后在 application.properties 中配置一下 RabbitMQ 的根本连贯信息：

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

接下来咱们来配置两个音讯队列：一个一般队列，一个死信队列：

@Configuration
public class QueueConfig {
    public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name";
    public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name";
    public static final String JAVABOY_ROUTING_KEY = "javaboy_routing_key";
    public static final String DLX_QUEUE_NAME = "dlx_queue_name";
    public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name";
    public static final String DLX_ROUTING_KEY = "dlx_routing_key";

    /**
     * 死信队列
     * @return
     */
    @Bean
    Queue dlxQueue() {return new Queue(DLX_QUEUE_NAME, true, false, false);
    }

    /**
     * 死信交换机
     * @return
     */
    @Bean
    DirectExchange dlxExchange() {return new DirectExchange(DLX_EXCHANGE_NAME, true, false);
    }

    /**
     * 绑定死信队列和死信交换机
     * @return
     */
    @Bean
    Binding dlxBinding() {return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
                .with(DLX_ROUTING_KEY);
    }

    /**
     * 一般音讯队列
     * @return
     */
    @Bean
    Queue javaboyQueue() {Map<String, Object> args = new HashMap<>();
        // 设置音讯过期工夫
        args.put("x-message-ttl", 1000*10);
        // 设置死信交换机
        args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        // 设置死信 routing_key
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);
    }

    /**
     * 一般交换机
     * @return
     */
    @Bean
    DirectExchange javaboyExchange() {return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);
    }

    /**
     * 绑定一般队列和与之对应的交换机
     * @return
     */
    @Bean
    Binding javaboyBinding() {return BindingBuilder.bind(javaboyQueue())
                .to(javaboyExchange())
                .with(JAVABOY_ROUTING_KEY);
    }
}

这段配置代码尽管略长，不过原理其实简略。

  • 配置能够分为两组，第一组配置死信队列，第二组配置一般队列。每一组都由音讯队列、音讯交换机以及 Binding 三者组成。
  • 配置音讯队列时，为音讯队列指定死信队列，不相熟的小伙伴能够翻一下上篇文章，传送门：RabbitMQ 中的音讯会过期吗？。
  • 配置队列中的音讯过期工夫时，默认的工夫单位时毫秒。

接下来咱们为死信队列配置一个消费者，如下：

@Component
public class DlxConsumer {private static final Logger logger = LoggerFactory.getLogger(DlxConsumer.class);

    @RabbitListener(queues = QueueConfig.DLX_QUEUE_NAME)
    public void handle(String msg) {logger.info(msg);
    }
}

收到音讯后就将之打印进去。

这就完事了。

启动我的项目。

最初咱们在单元测试中发送一条音讯：

@SpringBootTest
class DelayQueueApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {System.out.println(new Date());
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME, QueueConfig.JAVABOY_ROUTING_KEY, "hello javaboy!");
    }

}

这个就没啥好说的了，就是一般的音讯发送，10 秒之后这条音讯会在死信队列的消费者中被打印进去。

3. 小结

好啦，这就是咱们用 RabbitMQ 做提早队列的两种思路～感兴趣的小伙伴能够试试哦～

正文完
java
发表至： java
2021-12-07
 0
Java-最常见-200-面试题全解析面试必备附答案
关于-Spring-JdbcTemplate-的一些总结
关于java:再见-Typora-这款开源的-Markdown-神器界面更炫酷逼格更高
关于java:Java多线程学习笔记二-相识篇