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>

封装工具类用到的包

<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.发送消息，同时设置消息id
rabbitTemplate.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.发送消息，同时设置消息id
rabbitTemplate.convertAndSend(exchange, routingKey, t, new DelayedMessageProcessor(delay), correlationData);
}
/**
* 根据exchange和routingKey 异步发送消息，并指定一个延迟时间
*
* @param exchange   交换机
* @param routingKey 路由KEY
* @param t          数据
* @param         数据类型
*/
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         数据类型
*/
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;
}
@Bean
public MessageConverter messageConverter(ObjectMapper mapper){
// 1.定义消息转换器
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(mapper);
// 2.配置自动创建消息id，用于识别不同消息
jackson2JsonMessageConverter.setCreateMessageIds(true);
return jackson2JsonMessageConverter;
}
/**
* 
消息处理失败的重试策略
* 本地重试失败后，消息投递到专门的失败交换机和失败消息队列：error.queue
*/
@Bean
@ConditionalOnClass(MessageRecoverer.class)
@ConditionalOnMissingBean
public 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);
}
/**
* 专门接收处理失败的消息
*/
@Bean
public DirectExchange errorMessageExchange(){
return new DirectExchange(ERROR_EXCHANGE);
}
@Bean
public Queue errorQueue(){
return new Queue(defaultErrorQueue, true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(defaultErrorRoutingKey);
}
@Override
public 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";
@Override
public 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();
}
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 1.添加消息id
super.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-interval
max-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 {
@Autowired
RabbitMqHelper 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]