8

RabbitMQ-延迟队列

 2 years ago
source link: https://www.cnblogs.com/ludangxin/p/15302794.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
RabbitMQ-延迟队列 - 张铁牛 - 博客园

我们在上一篇博文中遗留了一个小问题，就是虽然TTL + DLX能实现延迟队列的功能，但是有两个问题。

首先业务场景为：比如海底捞预约，每个人预约的时间段不一致，有个可能一个小时后，有的可能三个小时等，当快到预约时间点需要给用户进行短信通知。

  1. 通过给Queue设置过期时间的方式不现实，因为很有可能每条记录的过期时间都不一样，不可能设置那么多的Queue。
  2. 直接给Message设置过期时间，这种方式也不好，因为这种方式是当该消息在队列头部时（消费时），才会单独判断这一消息是否过期。例：现在有两条消息，第一条消息过期时间为30s，而第二条消息过期时间为15s，当过了15秒后，第二条消息不会立即过期，而是要等第一条消息被消费后，第二条消息被消费时，才会判断是否过期，也就是等到第二条消息投往DLX已经过去45s了。

这也就抛出了本章主题：延迟队列。

RabbitMQ默认没有提供延迟队列功能，而是要通过插件提供的x-delayed-message（延迟交换机）来实现。

延迟队列：用户可以使用该类型声明一个交换，x-delayed-message然后使用自定义标头发布消息，x-delay以毫秒为单位表示消息的延迟时间。消息将在x-delay毫秒后传递到相应的队列。

2. 安装插件

官方插件地址：https://www.rabbitmq.com/community-plugins.html

找到插件rabbitmq_delayed_message_exchange，进入GitHub下载本地RabbitMQ对应的插件版本（下载.ez文件）。

我这里下载的是3.8.9版本，如图：

下载到本地后将文件放置RabbitMQ的plugins目录。

我这里本地是使用docker-compose安装的服务，image为rabbitmq:3.8.3-management（虽然版本没对起来，但是测试能用，但是使用3.9的版本会报错，插件安装失败）安装的服务，操作步骤如下：

  1. 将下载好的文件放置RabbitMQ插件目录

    rabbitmq：容器服务名

    $ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/

  2. $ docker exec -it rabbitmq /bin/bash

  3. 查看现有的插件列表

    $ rabbitmq-plugins list
# 输出部分内容如下 [E*] = 明确启用; e = 隐式启用
[  ] rabbitmq_amqp1_0                  3.8.3
[  ] rabbitmq_auth_backend_cache       3.8.3
[  ] rabbitmq_auth_backend_http        3.8.3
[  ] rabbitmq_auth_backend_ldap        3.8.3
[  ] rabbitmq_auth_backend_oauth2      3.8.3
[  ] rabbitmq_auth_mechanism_ssl       3.8.3
[  ] rabbitmq_consistent_hash_exchange 3.8.3
[  ] rabbitmq_event_exchange           3.8.3
[  ] rabbitmq_federation               3.8.3
[  ] rabbitmq_federation_management    3.8.3
[  ] rabbitmq_jms_topic_exchange       3.8.3
[E*] rabbitmq_management               3.8.3
[e*] rabbitmq_management_agent         3.8.3
[  ] rabbitmq_mqtt                     3.8.3

  4. $ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

  5. 再次查看安装列表就有了rabbitmq_delayed_message_exchange

安装完毕后登陆RabbitMQ控制台查看，会发现多了个x-delayed-message类型的Exchange。

3. 实现延迟队列

3.1 引入所需依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>

3.2 application.yaml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    # rabbit 默认的虚拟主机
    virtual-host: /
    # rabbit 用户名密码
    username: admin
    password: admin123

3.3 RabbitConfig

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

/**
 * 延迟队列配置
 *
 * @author ludangxin
 * @date 2021/9/16
 */
@Configuration
public class RabbitDelayedConfig {
    public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE";
    public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE";
    public static final String ROUTING_KEY_DELAYED = "DELAY.#";

    @Bean(QUEUE_NAME_DELAYED)
    public Queue queue() {
       return QueueBuilder.durable(QUEUE_NAME_DELAYED).build();
    }

    @Bean(EXCHANGE_NAME_DELAYED)
    public CustomExchange exchange() {
       Map<String, Object> arguments = new HashMap<>(1);
       // 在这里声明一个主题类型的延迟队列，当然其他类型的也可以。
       arguments.put("x-delayed-type", "topic");
       return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message", true, false, arguments);
    }

    @Bean
    public Binding bindingNotify(@Qualifier(QUEUE_NAME_DELAYED) Queue queue, @Qualifier(EXCHANGE_NAME_DELAYED) CustomExchange customExchange) {
       return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs();
    }
}

3.4 Producer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延迟消息生产者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Component
public class DelayProducer {

   @Autowired
   private RabbitTemplate rabbitTemplate;

   public void sendDelayedMsg(String msg, Integer delay) {
      MessageProperties mp = new MessageProperties();
      // 设置过期时间
      mp.setDelay(delay);
      Message message = new Message(msg.getBytes(), mp);
      rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG", message);
   }
}

3.5 Consumer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延迟消息消费者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Slf4j
@Component
public class DelayConsumer {

    @RabbitListener(queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED})
    public void delayQueue(Message message){
        log.info(new String(message.getBody()) + "，结束时间为：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    }

}

3.6 测试代码

@Autowired
private DelayProducer delayProducer;

@Test
@SneakyThrows
public void sendDelayedMsg() {
   for(int i = 16; i >= 10; i --) {
      String msg = "我将在" + i + "s后过期，开始时间为：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
      delayProducer.sendDelayedMsg(msg,i * 1000);
   }
   // 使进程阻塞，方便Consumer监听输出Message
   System.in.read();
}

3.7 启动测试

启动测试代码，输出内容如下：

从日志内容可以看出，消息存活了30s，符合预期。

2021-09-16 23:40:10.806  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在10s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:10
2021-09-16 23:40:11.792  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在11s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:11
2021-09-16 23:40:12.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在12s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:12
2021-09-16 23:40:13.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在13s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:13
2021-09-16 23:40:14.788  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在14s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:14
2021-09-16 23:40:15.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在15s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:15
2021-09-16 23:40:16.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在16s后过期，开始时间为：2021-09-16 23:40:00，结束时间为：2021-09-16 23:40:16

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK