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

1. Get-Started

docker安装rabbitmq

拉取镜像

[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

mkdir -p /home/apps/rabbitmq/data

运行容器

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

启用web界面管理插件

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

浏览器访问 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.发送消息，同时设置消息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 <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;
    }

    @Bean
    public 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)
    @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]