⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 blog.csdn.net/m0_55712478/article/details/135242345 「入秋的大橘」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

1 为什么要使用MQ？

在Spring Boot Event这篇文章中已经通过Guava或者SpringBoot自身的Listener实现了事件驱动，已经做到了对业务的解耦。为什么还要用到MQ来进行业务解耦呢？

首先无论是通过Guava还是Spring Boot自身提供的监听注解来实现的事件驱动他都是处于同一进程中的，意思就是当前事件推送后只有当前的进程可以进行消费。

通过MQ可以实现将事件推送到进程外的Broker中，在多实例/分布式环境下，其他的服务在订阅同一事件（Topic）时，可以在各自的服务中进行消费，最大化空闲服务的利用。

源码地址：

https://gitee.com/sparkle3021/springboot3-study

2 整合RocketMQ

依赖版本

JDK 17
Spring Boot 3.2.0
RocketMQ-Client 5.0.4
RocketMQ-Starter 2.2.0

Spring Boot 3.0+ 取消了对spring.factories的支持。所以在导入时需要手动引入RocketMQ的配置类。

引入RocketMQ依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>5.0.4</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version>
</dependency>

解决Spring Boot3+不兼容 spring.factories

rocketmq-spring-boot-starter:2.2.2版本中：

参考配置文件

# RocketMQ 配置
rocketmq:
  name-server: 127.0.0.1:9876
  consumer:
    group: event-mq-group
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 1
  producer:
    # 发送同一类消息的设置为同一个group，保证唯一
    group: event-mq-group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 10000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false

参考Issue

方法一：通过@Import(RocketMQAutoConfiguration.class)在配置类中引入
方法二：在resources资源目录下创建文件夹及文件META-INF/spring，org.springframework.boot.autoconfigure.AutoConfiguration.imports。

文件内容为RocketMQ自动配置类路径：org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration

3 RocketMQ 使用

解决Spring Boot3+不支持spring.factories的问题

import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Import;

/**
 * 启动类
 */
@Import(RocketMQAutoConfiguration.class)
@SpringBootApplication
public class MQEventApplication {
    public static void main(String[] args) {
        SpringApplication.run(MQEventApplication.class, args);
    }
}

RocketMQ操作工具

RocketMQ Message实体

import cn.hutool.core.util.IdUtil;
import jakarta.validation.constraints.NotBlank;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.io.Serializable;
import java.util.List;

/**
 * RocketMQ 消息
 */
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class RocketMQMessage<T> implements Serializable {

    /**
     * 消息队列主题
     */
    @NotBlank(message = "MQ Topic 不能为空")
    private String topic;

    /**
     * 延迟级别
     */
    @Builder.Default
    private DelayLevel delayLevel = DelayLevel.OFF;

    /**
     * 消息体
     */
    private T message;

    /**
     * 消息体
     */
    private List<T> messages;

    /**
     * 使用有序消息发送时，指定发送到队列
     */
    private String hashKey;

    /**
     * 任务Id，用于日志打印相关信息
     */
    @Builder.Default
    private String taskId = IdUtil.fastSimpleUUID();
}

RocketMQTemplate 二次封装

import com.yiyan.study.domain.RocketMQMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * RocketMQ 消息工具类
 */
@Slf4j
@Component
public class RocketMQService {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.sendMessageTimeout}")
    private int sendMessageTimeout;

    /**
     * 异步发送消息回调
     *
     * @param taskId 任务Id
     * @param topic  消息主题
     * @return the send callback
     */
    private static SendCallback asyncSendCallback(String taskId, String topic) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("ROCKETMQ 异步消息发送成功 : [TaskId:{}] - [Topic:{}] - [SendStatus:{}]", taskId, topic, sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("ROCKETMQ 异步消息发送失败 : [TaskId:{}] - [Topic:{}] - [ErrorMessage:{}]", taskId, topic, throwable.getMessage());
            }
        };
    }

    /**
     * 发送同步消息，使用有序发送请设置HashKey
     *
     * @param message 消息参数
     */
    public <T> void syncSend(RocketMQMessage<T> message) {
        log.info("ROCKETMQ 同步消息发送 : [TaskId:{}] - [Topic:{}]", message.getTaskId(), message.getTopic());
        SendResult sendResult;
        if (StringUtils.isNotBlank(message.getHashKey())) {
            sendResult = rocketMQTemplate.syncSendOrderly(message.getTopic(), message.getMessage(), message.getHashKey());
        } else {
            sendResult = rocketMQTemplate.syncSend(message.getTopic(), message.getMessage(), sendMessageTimeout, message.getDelayLevel().getLevel());
        }
        log.info("ROCKETMQ 同步消息发送结果 : [TaskId:{}] - [Topic:{}] - [MessageId:{}] - [SendStatus:{}]",
                message.getTaskId(), message.getTopic(), sendResult.getMsgId(), sendResult.getSendStatus());
    }

    /**
     * 批量发送同步消息
     *
     * @param message 消息参数
     */
    public <T> void syncSendBatch(RocketMQMessage<T> message) {
        log.info("ROCKETMQ 同步消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount:{}]",
                message.getTaskId(), message.getTopic(), message.getMessages().size());
        SendResult sendResult;
        if (StringUtils.isNotBlank(message.getHashKey())) {
            sendResult = rocketMQTemplate.syncSendOrderly(message.getTopic(), message.getMessages(), message.getHashKey());
        } else {
            sendResult = rocketMQTemplate.syncSend(message.getTopic(), message.getMessages());
        }
        log.info("ROCKETMQ 同步消息-批量发送结果 : [TaskId:{}] - [Topic:{}] - [MessageId:{}] - [SendStatus:{}]",
                message.getTaskId(), message.getTopic(), sendResult.getMsgId(), sendResult.getSendStatus());
    }

    /**
     * 异步发送消息，异步返回消息结果
     *
     * @param message 消息参数
     */
    public <T> void asyncSend(RocketMQMessage<T> message) {
        log.info("ROCKETMQ 异步消息发送 : [TaskId:{}] - [Topic:{}]", message.getTaskId(), message.getTopic());
        if (StringUtils.isNotBlank(message.getHashKey())) {
            rocketMQTemplate.asyncSendOrderly(message.getTopic(), message.getMessage(), message.getHashKey(),
                    asyncSendCallback(message.getTaskId(), message.getTopic()));
        } else {
            rocketMQTemplate.asyncSend(message.getTopic(), message.getMessage(),
                    asyncSendCallback(message.getTaskId(), message.getTopic()), sendMessageTimeout, message.getDelayLevel().getLevel());
        }
    }

    /**
     * 批量异步发送消息
     *
     * @param message 消息参数
     */
    public <T> void asyncSendBatch(RocketMQMessage<T> message) {
        log.info("ROCKETMQ 异步消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount:{}]",
                message.getTaskId(), message.getTopic(), message.getMessages().size());
        if (StringUtils.isNotBlank(message.getHashKey())) {
            rocketMQTemplate.asyncSendOrderly(message.getTopic(), message.getMessages(), message.getHashKey(),
                    asyncSendCallback(message.getTaskId(), message.getTopic()));
        } else {
            rocketMQTemplate.asyncSend(message.getTopic(), message.getMessages(),
                    asyncSendCallback(message.getTaskId(), message.getTopic()));
        }
    }

    /**
     * 单向发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送;
     *
     * @param message 消息参数
     */
    public <T> void sendOneWay(RocketMQMessage<T> message) {
        sendOneWay(message, false);
    }

    /**
     * 单向消息 - 批量发送
     *
     * @param message 消息体
     * @param batch   是否为批量操作
     */
    public <T> void sendOneWay(RocketMQMessage<T> message, boolean batch) {
        log.info((batch ? "ROCKETMQ 单向消息发送 : [TaskId:{}] - [Topic:{}]"
                        : "ROCKETMQ 单向消息-批量发送 : [TaskId:{}] - [Topic:{}] - [MessageCount{}]"),
                message.getTaskId(), message.getTopic(), message.getMessages().size());
        if (StringUtils.isNotBlank(message.getHashKey())) {
            if (batch) {
                message.getMessages().
                        forEach(msg -> rocketMQTemplate.sendOneWayOrderly(message.getTopic(), msg, message.getHashKey()));
            } else {
                rocketMQTemplate.sendOneWayOrderly(message.getTopic(), message.getMessage(), message.getHashKey());
            }
        } else {
            if (batch) {
                message.getMessages().forEach(msg -> rocketMQTemplate.sendOneWay(message.getTopic(), msg));
            } else {
                rocketMQTemplate.sendOneWay(message.getTopic(), message.getMessage());
            }
        }
    }
}

定义RocketMQ消费者

import com.yiyan.study.constants.MQConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * MQ消息监听
 */
@Component
@Slf4j
@RocketMQMessageListener(topic = MQConfig.EVENT_TOPIC,
        consumerGroup = MQConfig.EVENT_CONSUMER_GROUP)
public class MQListener implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        log.info("MQListener 接收消息 ： {}", message);
    }
}

定义测试类发送消息

import cn.hutool.core.thread.ThreadUtil;
import com.yiyan.study.constants.MQConfig;
import com.yiyan.study.domain.RocketMQMessage;
import com.yiyan.study.utils.RocketMQService;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * MQ测试
 */
@SpringBootTest
public class MQTest {

    @Resource
    private RocketMQService rocketMQService;

    @Test
    public void sendMessage() {
        int count = 1;
        while (count <= 50) {
            rocketMQService.syncSend(RocketMQMessage.builder()
                    .topic(MQConfig.EVENT_TOPIC)
                    .message(count++)
                    .build());
        }
        // 休眠等待消费消息
        ThreadUtil.sleep(2000L);
    }
}

芋道源码 —— 纯源码解析博客

Spring Boot+RocketMQ 实现多实例分布式环境下的事件驱动