消息队列

这篇解决什么

短信、邮件、跨模块回写不必堵在 HTTP 返回之前。读完能选 Spring Event、Redis Stream / Pub/Sub，或打开 RocketMQ / Kafka / RabbitMQ，并接上 Producer 与 Consumer。

默认端口 48080，管理端前缀 /admin-api。短信和邮件默认走进程内 Event。要落盘、重启不丢、多实例只消费一次，用 Redis Stream。

示意图：业务模块只发消息、写监听器；框架按你选的后端投递。

怎么选

单进程旁路（短信、邮件）用 Event + @Async。要堆积、只消费一次，用 Redis Stream（需 Redis 5.0+）。每个实例都要收到（WebSocket 推送）用 Pub/Sub 或其它中间件广播。机房已有 RocketMQ / Kafka / Rabbit 时，打开 optional 依赖，用官方 Template。

组件位置

名称	说明	仓库路径
MQ Starter	Redis 封装 + 可选 RocketMQ / Kafka / Rabbit	`ruoyi-office/yudao-framework/yudao-spring-boot-starter-mq/`
自动配置	Redis 生产/消费、Rabbit JSON 转换	`.../META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports`
Redis 发送	`RedisMQTemplate`	`.../mq/redis/core/RedisMQTemplate.java`
全局连接	Producer 分组、Kafka 序列化	`ruoyi-office/yudao-server/src/main/resources/application.yaml`
本机地址	Namesrv `9876`、Rabbit `5672`、Kafka `9092`	`.../application-local.yaml`
BOM	`rocketmq-spring-boot-starter` `2.3.6`	`ruoyi-office/yudao-dependencies/pom.xml`

system、infra、bpm、oa、合同 / HRM / 项目 / CRM 等 *-server 已引入 Starter。新模块先加这个依赖，不要自己再写一套 Template。

名称	说明	仓库路径
短信 Event	发完日志再异步调渠道	`yudao-module-system-server/.../mq/producer/sms/SmsProducer.java`
邮件 Event	同上	`.../mq/producer/mail/MailProducer.java`
合同 Stream	生效 / 终止后通知其它模块	`yudao-module-contract-server/.../mq/producer/ContractStatusChangedProducer.java`
WebSocket 广播	`sender-type`：`local` / `redis` / `rocketmq` / `kafka` / `rabbitmq`	`yudao-spring-boot-starter-websocket/`
IoT 总线	`yudao.iot.message-bus.type`，缺省 `local`	`yudao-module-iot-core/.../IotMessageBusAutoConfiguration.java`
BPM 通知	开箱只有 Redis Stream	`yudao-module-bpm-server/.../application-notification.yaml`

Spring Event

短信、邮件默认不接外部 Broker。Producer 往 ApplicationContext 发事件，Consumer 用 @EventListener + @Async 离开请求线程。

名称	说明	仓库路径
消息体	`logId`、手机号、渠道、模板、参数	`.../mq/message/sms/SmsSendMessage.java`
生产者	`publishEvent`	`SmsProducer`
消费者	`@EventListener` + `@Async`	`SmsSendConsumer`
真正发送	调渠道 API、回写日志	`SmsSendService.doSendSms`
邮件一套	`MailSendMessage` / `MailProducer` / `MailSendConsumer`	`.../mq/` 下 `mail` 包

java

public void sendSmsSendMessage(Long logId, String mobile,
                               Long channelId, String apiTemplateId,
                               List<KeyValue<String, Object>> templateParams) {
    SmsSendMessage message = new SmsSendMessage().setLogId(logId).setMobile(mobile);
    message.setChannelId(channelId).setApiTemplateId(apiTemplateId)
            .setTemplateParams(templateParams);
    applicationContext.publishEvent(message);
}

@EventListener
@Async
public void onMessage(SmsSendMessage message) {
    smsSendService.doSendSms(message);
}

@Async 的线程池、租户传递见异步任务。同类内部调用必须走 Spring 代理，不能 this.onMessage(...)。

重启会丢掉未跑完的 Event

消息只在当前 JVM 堆里。进程停掉，还没被 @Async 执行的短信 / 邮件不会重投。要堆积、要多实例抢一条，改 Redis Stream 或 RocketMQ。

本地打断点：请求在 http-nio-*，消费在 task-*。日志出现 [onMessage][消息内容 即发送和消费都通了。

Redis 集群消费

集群消费：一条消息只会被一个 JVM 实例拿到。短信要可靠堆积时可以改走这条；合同状态变更已经在用。

实现是 Redis Stream。YudaoRedisMQConsumerAutoConfiguration 启动时扫描全部 AbstractRedisStreamMessageListener，建消费组，再 XREADGROUP。

名称	说明	仓库路径
消息基类	默认 Stream Key = 类名	`.../stream/AbstractRedisStreamMessage.java`
监听基类	反序列化、拦截器、成功才 ACK	`.../stream/AbstractRedisStreamMessageListener.java`
发送	`opsForStream().add`	`RedisMQTemplate.send`
消费组	默认 `spring.application.name`	监听器上的 `group`
消费者名	`IP@PID`	`YudaoRedisMQConsumerAutoConfiguration.buildConsumerName`
版本门槛	主版本 < 5 直接拒绝启动	`checkRedisVersion`

示意图：写入 Stream；一个消费组里只有一个消费者处理；失败留下 pending。

消费失败必须吞掉异常。抛给容器会让整条 Stream 进入 1～30 秒退避，后面的用户变更、主数据、流程回调一起堵。不 ACK，交给 pending 任务重试。

名称	说明	仓库路径
重投	每分钟第 35 秒扫超时 5 分钟的 pending	`RedisPendingMessageResendJob`
清理	`XTRIM` 只留最近 10000 条	`RedisStreamMessageCleanupJob`
锁	多实例互斥，IoT 用另一把锁	`redis:stream:pending-message-resend:lock`

合同状态变更

合同模块只发布，不反向依赖 HRM。HRM 按固定 Stream Key 回写员工合同快照。

名称	说明	仓库路径
消息	Key `contract.status-changed`	`yudao-module-contract-api/.../ContractStatusChangedMessage.java`
生产者	事务提交后再发	`ContractStatusChangedProducer`
消费者	只处理员工相对方	`yudao-module-hrm-server/.../HrmContractStatusChangedConsumer.java`

java

@Data
public class ContractStatusChangedMessage extends AbstractRedisStreamMessage {
    public static final String STREAM_KEY = "contract.status-changed";
    private ContractStatusChangedPayload data;
    @Override
    public String getStreamKey() {
        return STREAM_KEY;
    }
}

redisMQTemplate.send(new ContractStatusChangedMessage().setData(payload));

@Component
public class HrmContractStatusChangedConsumer
        extends AbstractRedisStreamMessageListener<ContractStatusChangedMessage> {
    @Override
    public void onMessage(ContractStatusChangedMessage message) {
        employeeContractService.refreshSnapshot(employeeId);
    }
}

自己加一条 Stream：消息继承 AbstractRedisStreamMessage（需要稳定 Key 就覆盖 getStreamKey），Producer 调 redisMQTemplate.send，Consumer 继承 AbstractRedisStreamMessageListener<T> 并注册成 Bean。监听器泛型必须写死，解析不到消息类型会启动失败。

先提交事务再发 Stream

ContractStatusChangedProducer 挂在 afterCommit。事务还没提交就发，消费者查库会读到旧行。afterCommit 时线程上的租户可能已空，要按载荷 TenantUtils.execute 再发送，Header 里才有 tenant-id。

OA 用户变更、项目 / CRM / MES 主数据同步，同一套路：*-api 里放消息类，*-server 里放 Producer / Listener。

Redis 广播消费

广播消费：每个订阅了该 Channel 的 JVM 都会收到。刷新进程内缓存、把 WebSocket 推到「用户连在哪台机器」时用它。

实现是 Redis Pub/Sub。RedisMQTemplate.send(AbstractRedisChannelMessage) 走 convertAndSend。YudaoRedisMQConsumerAutoConfiguration 扫描 AbstractRedisChannelMessageListener，挂到 RedisMessageListenerContainer。

名称	说明	仓库路径
消息基类	默认 Channel = 类名	`.../pubsub/AbstractRedisChannelMessage.java`
监听基类	反序列化后再 `onMessage`	`.../pubsub/AbstractRedisChannelMessageListener.java`
WebSocket 帧	Session / 用户 / `type` / 内容	`.../websocket/core/sender/redis/RedisWebSocketMessage.java`
WebSocket 消费	收到后再查本机 Session	`RedisWebSocketMessageConsumer`

示意图：频道上每个进程都消费；没有该用户连接的实例直接跳过。

多实例 WebSocket 把 yudao.websocket.sender-type 改成 redis。默认 local 只扫本进程 Session，连在别的节点上的用户收不到。细节见 WebSocket。

Pub/Sub 不落盘。没人订阅时发出去就丢。要堆积仍用 Stream。

RocketMQ、Kafka、RabbitMQ

Starter 的 pom.xml 里三套客户端都是 <optional>true</optional>。只配 application-local.yaml 的地址，不会自动连上。要在引入 Starter 的模块（通常 yudao-module-system-server 或 yudao-server）里再声明对应依赖，并刷新 Maven。

名称	说明	仓库路径
RocketMQ	`rocketmq-spring-boot-starter`，同步 `RocketMQTemplate`	Starter `pom.xml`
Kafka	`spring-kafka`，`KafkaTemplate` / `@KafkaListener`	同上
RabbitMQ	`spring-rabbit`，`RabbitTemplate` / `@RabbitListener`	同上
Rabbit JSON	`Jackson2JsonMessageConverter`	`YudaoRabbitMQAutoConfiguration`
Namesrv	`rocketmq.name-server: 127.0.0.1:9876`	`application-local.yaml`
Rabbit 本机	`host/port`，账号 `rabbit` / `rabbit`	同上
Kafka Broker	`spring.kafka.bootstrap-servers: 127.0.0.1:9092`	同上
全局 Kafka	`acks=1`、`JsonSerializer`、`missing-topics-fatal=false`	`application.yaml`

application.yaml 已写 rocketmq.producer.group: ${spring.application.name}_PRODUCER。单体里应用名是 yudao-server。

没声明依赖就没有 Template

RocketMQ / Kafka / Rabbit 的自动配置都挂在 ConditionalOnClass 上。classpath 没有客户端时，WebSocket 的 sender-type、IoT 的 message-bus.type 也切不过去。

把短信从 Event 改到 RocketMQ 时，消息体加 Topic，Producer 换 RocketMQTemplate，Consumer 实现 RocketMQListener：

java

public class SmsSendMessage {
    public static final String TOPIC = "SMS_SEND_TOPIC";
    // logId / mobile / channelId / apiTemplateId / templateParams
}

rocketMQTemplate.syncSend(SmsSendMessage.TOPIC, message);

@Component
@RocketMQMessageListener(
        topic = SmsSendMessage.TOPIC,
        consumerGroup = SmsSendMessage.TOPIC + "_CONSUMER")
public class SmsSendConsumer implements RocketMQListener<SmsSendMessage> {
    @Override
    public void onMessage(SmsSendMessage message) {
        smsSendService.doSendSms(message);
    }
}

Kafka 同一套：消息实现 Serializable，Topic 例如 SEND_MESSAGE_TOPIC，kafkaTemplate.send，方法上 @KafkaListener(topics, groupId)。

Rabbit 同一套：消息实现 Serializable，Queue 例如 SEND_MESSAGE_QUEUE，控制台先建同名队列，rabbitTemplate.convertAndSend，类上 @RabbitListener，方法上 @RabbitHandler。

WebSocket 多实例也可以不走 Redis：yudao.websocket.sender-type 设为 rocketmq / kafka / rabbitmq。RocketMQ 消费者显式 MessageModel.BROADCASTING，保证每台机器都收到再查本地 Session。

IoT 设备报文走独立总线：yudao.iot.message-bus.type 可选 local（缺省，进程内 Event）、redis、rocketmq、kafka、rabbitmq。和短信不是同一条 Topic。

BPM 流程通知开箱只有 Redis Stream（yudao.bpm.notification.mq.type=redis）。yaml 里预留了其它类型名，但没有对应发送器，不要只改字符串。

租户

发送线程上的 TenantContextHolder 必须写进消息头，消费线程再还原。Header 名是 tenant-id，和 HTTP 一致。

名称	说明	仓库路径
Redis	写入 `AbstractRedisMessage.headers`	`TenantRedisMessageInterceptor`
RocketMQ	`putUserProperty`，消费 Hook 还原	`TenantRocketMQSendMessageHook` / `ConsumeMessageHook`
Rabbit	`MessageProperties` Header	`TenantRabbitMQMessagePostProcessor`
Kafka	Producer Header；消费靠 `InvocableHandlerMethod`	`TenantKafkaProducerInterceptor`

Redis 拦截器在 consumeMessageAfter 里 TenantContextHolder.clear()。消费方法里不要依赖「结束后租户还在」。

Event 没有消息头。短信 / 邮件靠 @Async 的 TTL 把租户拷到工作线程。手搓 Executors 会丢上下文，见异步任务篇。

消费线程不要假定租户还在

afterCommit、定时任务、忽略租户的调用里，getTenantId() 经常是空的。Stream 消息把租户放进载荷，发送前 TenantUtils.execute；消费时再按载荷包一层。

新业务优先 Redis Stream：本地已经有 Redis，Starter 会建组、重投、裁剪。机房统一用 RocketMQ / Kafka 时再打开 optional 依赖，按上面改 Producer / Consumer。不要同时往两套 Broker 发同一条业务消息。

配置与操作

没有消息队列菜单。改 yudao-server 的 yaml（rocketmq、yudao.websocket.sender-type）并打开对应 optional 依赖，重启后 Producer / Consumer 才换通道。

新业务优先 Redis Stream。不要同时往两套 Broker 发同一条业务消息。开启见基建功能设计。

消息队列 ​

这篇解决什么 ​

组件位置 ​

Spring Event ​

Redis 集群消费 ​

合同状态变更 ​

Redis 广播消费 ​

RocketMQ、Kafka、RabbitMQ ​

租户 ​

配置与操作 ​

相关篇 ​

消息队列

这篇解决什么

组件位置

Spring Event

Redis 集群消费

合同状态变更

Redis 广播消费

RocketMQ、Kafka、RabbitMQ

租户

配置与操作

相关篇