RabbitMQ安装配置，封装工具类，发送消息及监听

1. Get-Started

docker安装rabbitmq

1. 拉取镜像

[root@heima ~]# docker pull rabbitmq:3.8-management
3.8-management: Pulling from library/rabbitmq
7b1a6ab2e44d: Pull complete 
37f453d83d8f: Pull complete 
e64e769bc4fd: Pull complete 
c288a913222f: Pull complete 
13adc5da62c6: Pull complete 
bd67e639afcb: Pull complete 
9a48b5ad2519: Pull complete 
1cdfc59624be: Pull complete 
8f5ad79f0ad6: Pull complete 
Digest: sha256:543f7268600a27a39e2fdd532f8df479636fc0cf528aadde88d5fe718bed71e4
Status: Downloaded newer image for rabbitmq:3.8-management
docker.io/library/rabbitmq:3.8-management

2. 创建目录

mkdir -p /home/apps/rabbitmq/data

3. 运行容器

docker run \
-d \
--name rabbitmq \
--restart=always \
--privileged=true \
-p 5672:5672 \
-p 15672:15672 \
-v /home/apps/rabbitmq/data:/var/lib/rabbitmq \
-e RABBITMQ_DEFAULT_VHOST=vhost0 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin123 \
rabbitmq:3.8-management

4. 启用web界面管理插件

docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_management
或
-- 进入容器
docker exec -it rabbitmq /bin/bash
-- 安装插件
rabbitmq-plugins enable rabbitmq_management

4. 浏览器访问 http://虚拟机ip:15672/出现以下界面说明安装成功。

输入上面在初始化Rabbitmq容器时我们自己指定了默认账号和密码：admin/admin123，如果没有指定的话那么rabbitmq的默认账号密码是:guest/guest

扩展 ：安装延迟队列

• 去官网下载插件（v3.8.17）
• 地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

# 将下载好的插件复制到mq容器内部
docker cp rabbitmq_delayed_message_exchange-3.8.17.8f537ac.ez rabbitmq:/plugins# 进入mq容器
docker exec -it rabbitmq /bin/bash# 开启插件支持 
rabbitmq-plugins enable rabbitmq_delayed_message_exchange# 查看插件列表
rabbitmq-plugins list

2. 生产者模块

2.1 项目引入依赖：

rabbitmq依赖包，使用RabbitMq这2个依赖就够了。

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-amqp</artifactId><scope>provided</scope>
</dependency>
<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><scope>provided</scope>
</dependency>

封装工具类用到的包

<!--hutool工具包-->
<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.7.17</version>
</dependency>

2.2 工具类

• RabbitMqHelper

import cn.hutool.core.lang.UUID;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
@Slf4j
public class RabbitMqHelper {public static final String REQUEST_ID_HEADER = "requestId";private final RabbitTemplate rabbitTemplate;private final MessagePostProcessor processor = new BasicIdMessageProcessor();private final ThreadPoolTaskExecutor executor;public RabbitMqHelper(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;executor = new ThreadPoolTaskExecutor();//配置核心线程数executor.setCorePoolSize(10);//配置最大线程数executor.setMaxPoolSize(15);//配置队列大小executor.setQueueCapacity(99999);//配置线程池中的线程的名称前缀executor.setThreadNamePrefix("mq-async-send-handler");// 设置拒绝策略：当pool已经达到max size的时候，如何处理新任务// CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//执行初始化executor.initialize();}/*** 根据exchange和routingKey发送消息*/public <T> void send(String exchange, String routingKey, T t) {log.debug("准备发送消息，exchange：{}， RoutingKey：{}， message：{}", exchange, routingKey, t);// 1.设置消息标示，用于消息确认，消息发送失败直接抛出异常，交给调用者处理String id = UUID.randomUUID().toString(true);CorrelationData correlationData = new CorrelationData(id);// 2.设置发送超时时间为500毫秒rabbitTemplate.setReplyTimeout(500);// 3.发送消息，同时设置消息idrabbitTemplate.convertAndSend(exchange, routingKey, t, processor, correlationData);}/*** 根据exchange和routingKey发送消息，并且可以设置延迟时间*/public <T> void sendDelayMessage(String exchange, String routingKey, T t, Duration delay) {// 1.设置消息标示，用于消息确认，消息发送失败直接抛出异常，交给调用者处理String id = UUID.randomUUID().toString(true);CorrelationData correlationData = new CorrelationData(id);// 2.设置发送超时时间为500毫秒rabbitTemplate.setReplyTimeout(500);// 3.发送消息，同时设置消息idrabbitTemplate.convertAndSend(exchange, routingKey, t, new DelayedMessageProcessor(delay), correlationData);}/*** 根据exchange和routingKey 异步发送消息，并指定一个延迟时间** @param exchange   交换机* @param routingKey 路由KEY* @param t          数据* @param <T>        数据类型*/public <T> void sendAsync(String exchange, String routingKey, T t, Long time) {String requestId = MDC.get(REQUEST_ID_HEADER);CompletableFuture.runAsync(() -> {try {MDC.put(REQUEST_ID_HEADER, requestId);// 发送延迟消息if (time != null && time > 0) {sendDelayMessage(exchange, routingKey, t, Duration.ofMillis(time));} else {send(exchange, routingKey, t);}} catch (Exception e) {log.error("推送消息异常，t:{},", t, e);}}, executor);}/*** 根据exchange和routingKey 异步发送消息** @param exchange   交换机* @param routingKey 路由KEY* @param t          数据* @param <T>        数据类型*/public <T> void sendAsync(String exchange, String routingKey, T t) {sendAsync(exchange, routingKey, t, null);}
}

• MqConfig

import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.MDC;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;@Configuration
@ConditionalOnClass(value = {MessageConverter.class, AmqpTemplate.class})
public class MqConfig implements EnvironmentAware{public static final String REQUEST_ID_HEADER = "requestId";public static final String ERROR_EXCHANGE = "error.topic";public static final String ERROR_KEY_PREFIX = "error.";public static final String ERROR_QUEUE_TEMPLATE = "error.{}.queue";private String defaultErrorRoutingKey;private String defaultErrorQueue;@Bean(name = "rabbitListenerContainerFactory")@ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "simple",matchIfMissing = true)SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory,ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();configurer.configure(factory, connectionFactory);simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);factory.setAfterReceivePostProcessors(message -> {Object header = message.getMessageProperties().getHeader(REQUEST_ID_HEADER);if(header != null) {MDC.put(REQUEST_ID_HEADER, header.toString());}return message;});return factory;}@Beanpublic MessageConverter messageConverter(ObjectMapper mapper){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(mapper);// 2.配置自动创建消息id，用于识别不同消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;}/*** <h1>消息处理失败的重试策略</h1>* 本地重试失败后，消息投递到专门的失败交换机和失败消息队列：error.queue*/@Bean@ConditionalOnClass(MessageRecoverer.class)@ConditionalOnMissingBeanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){// 消息处理失败后，发送到错误交换机：error.direct，RoutingKey默认是error.微服务名称return new RepublishMessageRecoverer(rabbitTemplate, ERROR_EXCHANGE, defaultErrorRoutingKey);}/*** rabbitmq发送工具**/@Bean@ConditionalOnMissingBean@ConditionalOnClass(RabbitTemplate.class)public RabbitMqHelper rabbitMqHelper(RabbitTemplate rabbitTemplate){return new RabbitMqHelper(rabbitTemplate);}/*** 专门接收处理失败的消息*/@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange(ERROR_EXCHANGE);}@Beanpublic Queue errorQueue(){return new Queue(defaultErrorQueue, true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(defaultErrorRoutingKey);}@Overridepublic void setEnvironment(Environment environment) {String appName = environment.getProperty("spring.application.name");this.defaultErrorRoutingKey = ERROR_KEY_PREFIX + appName;this.defaultErrorQueue = StrUtil.format(ERROR_QUEUE_TEMPLATE, appName);}
}

• Processor

import cn.hutool.core.lang.UUID;
import org.slf4j.MDC;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;public class BasicIdMessageProcessor implements MessagePostProcessor {public static final String REQUEST_ID_HEADER = "requestId";@Overridepublic Message postProcessMessage(Message message) throws AmqpException {String requestId = MDC.get(REQUEST_ID_HEADER);if (requestId == null) {requestId = UUID.randomUUID().toString(true);}// 写入RequestID标示message.getMessageProperties().setHeader(REQUEST_ID_HEADER, requestId);return message;}
}----------------------------------------------------------------------import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import java.time.Duration;public class DelayedMessageProcessor extends BasicIdMessageProcessor {private final long delay;public DelayedMessageProcessor(Duration delay) {this.delay = delay.toMillis();}@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 1.添加消息idsuper.postProcessMessage(message);// 2.添加延迟时间message.getMessageProperties().setHeader("x-delay", delay);return message;}
}

2.3 配置

通常mq是放在common模块中，别的模块需要mq时就引入该模块，因此需要把MqConfig放在IOC容器里加载。

• resources/META-INF/spring.factories

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\com.tianji.common.autoconfigure.mq.MqConfig

配置rabbitmq的地址等信息

• boostrap.yml

spring:rabbitmq:host: ${mydemo.mq.host:192.168.150.101}port: ${mydemo.mq.port:5672}virtual-host: ${mydemo.mq.vhost:/vhost0}username: ${mydemo.mq.username:admin}password: ${mydemo.mq.password:admin123}listener:simple:retry:enabled: ${mydemo.mq.listener.retry.enable:true} # 开启消费者失败重试initial-interval: ${mydemo.mq.listener.retry.interval:1000ms} # 初始的失败等待时长为1秒multiplier: ${mydemo.mq.listener.retry.multiplier:2} # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: ${mydemo.mq.listener.retry.max-attempts:3} # 最大重试次数stateless: ${mydemo.mq.listener.retry.stateless:true} # true无状态；false有状态。如果业务中包含事务，这里改为false

2.4 测试文件

TestController

@RestController
public class TestController {@AutowiredRabbitMqHelper rabbitMqHelper;@GetMapping("/hello")public String hello(){return "hello";}@GetMapping("/sendMsg")public String sendMsg(){rabbitMqHelper.send("order.topic", // "order.topic""order.pay", // "order.pay"UserDto.builder().id(10001).name("gz").age(23).phone("15500000001").email("123@qq.com").build());return "success";}
}

UserDto

import lombok.Builder;
import lombok.Data;@Data
@Builder
public class UserDto {private Integer id;private String name;private String phone;private Integer age;private String email;
}

2.5 生产者测试

生产者模块目录结构

启动生产者服务，访问http://127.0.0.1:8080/sendMsg
报错了，原因是没有消费者消费消息
但是发现创建了新的错误消息exchange和queue

因此，接下来创建消费者监听消息。

3. 消费者

3.1 引入通用模块

由于消费者也需要使用刚刚的工具类和UserDto用来接收消息。

而这些工具类和UserDto都在模块rabbitmq-demo中。因此，在消费者模块中引入该生产者模块（如果这些通用的配置和实体类dto都在一个通用的模块common中，哪些模块需要发送消息，哪些模块需要监听消息，就都引入common模块就行了）

    <groupId>com.gzdemo</groupId><artifactId>rabbitmq-demo</artifactId><version>1.0-SNAPSHOT</version>

因为rabbitmq-demo模块中已经引入了RabbitMq相关依赖，因此消费者模块不需要重复引入RabbitMq相关的依赖。只需要把依赖包添加到classpath下,再import就行了

3.2 配置yml

与生产者的配置相同，通常只有消费者才需要配置listener，生产者不需要。

3.3 编写listener

import com.gzdemo.rabbitmq.pojos.UserDto;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class UserListener {/*** 监听消息*/@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "learning.lesson.pay.queue", durable = "true"),exchange = @Exchange(name = "order.topic", type = ExchangeTypes.TOPIC),key = "order.pay"))public void listenLessonPay(UserDto userDto){System.out.println(userDto);}
}

3.4 消费者目录结构

这里是随便找了个同工程中的子模块作为消费者测试，在UserListener类中监听消息

4 测试消息发送、监听

启动这两个服务，浏览器访问http://127.0.0.1:8080/sendMsg，多访问几次后发现消费者接收到了消息
生产者console

2024-06-14 05:46:39.187  INFO 106168 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [192.168.150.101:5672]
2024-06-14 05:46:39.210  INFO 106168 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#399d82:0/SimpleConnection@1673b17 [delegate=amqp://admin@192.168.150.101:5672/vhost0, localPort= 61317]