手把手带你掌握SpringBoot RabbitMQ延迟队列

手把手带你掌握SpringBoot RabbitMQ延迟队列

目录1. 简介2. 安装插件3. 实现延迟队列3.1 引入所需依赖3.2 application.yaml3.3 RabbitConfig3.4 Producer3.5 Consumer3.6 测试代码3.7 启动测试

1. 简介

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

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

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

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

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

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

2. 安装插件

官方插件地址：https://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

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

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

3. 实现延迟队列

3.1 引入所需依赖

org.springframework.boot

spring-boot-starter-amqp

org.projectlombok

lombok

true

org.springframework.boot

spring-boot-starter-test

test

org.springframework.amqp

spring-rabbit-test

test

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 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.sprNPMMJyfJvZingframework.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 DelaNPMMJyfJvZyProducer {

@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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。