diff --git a/ruoyi-admin/src/main/resources/application.yml b/ruoyi-admin/src/main/resources/application.yml
index 32b8a7429..3a07c5cb9 100644
--- a/ruoyi-admin/src/main/resources/application.yml
+++ b/ruoyi-admin/src/main/resources/application.yml
@@ -217,7 +217,7 @@ management:
--- # 统一消息推送配置
message:
enabled: true
- # sse / websocket
+ # sse / websocket / mqtt
transport: sse
# 统一访问路径
path: /resource/message
@@ -231,6 +231,8 @@ message:
web-socket-send-time-limit: 10000
# WebSocket 发送缓冲区大小
web-socket-buffer-size-limit: 64000
+ # MQTT 消息推送主题前缀,广播主题: ${prefix}/all,用户主题: ${prefix}/user/{userId}
+ mqtt-topic-prefix: /message
--- # warm-flow工作流配置
warm-flow:
diff --git a/ruoyi-common/ruoyi-common-push/pom.xml b/ruoyi-common/ruoyi-common-push/pom.xml
index f86a27f3a..98d143e0a 100644
--- a/ruoyi-common/ruoyi-common-push/pom.xml
+++ b/ruoyi-common/ruoyi-common-push/pom.xml
@@ -52,5 +52,10 @@
+
+
+ org.dromara
+ ruoyi-common-mqtt
+
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageMqttConfiguration.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageMqttConfiguration.java
new file mode 100644
index 000000000..d3ba9e35e
--- /dev/null
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageMqttConfiguration.java
@@ -0,0 +1,34 @@
+package org.dromara.common.push.config;
+
+import org.dromara.common.mqtt.config.MqttAutoConfiguration;
+import org.dromara.common.push.annotation.ConditionalOnMessageTransport;
+import org.dromara.common.push.core.MqttPushSessionManager;
+import org.dromara.common.push.properties.MessageProperties;
+import org.dromara.mica.mqtt.spring.client.MqttClientTemplate;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * MQTT 消息推送自动装配。
+ *
+ * @author Lion Li
+ */
+@AutoConfiguration(after = {MessageAutoConfiguration.class, MqttAutoConfiguration.class})
+@ConditionalOnMessageTransport("mqtt")
+@ConditionalOnProperty(value = "mqtt.client.enabled", havingValue = "true")
+public class MessageMqttConfiguration {
+
+ /**
+ * MQTT 推送会话管理器。
+ *
+ * @param client MQTT 客户端模板
+ * @param messageProperties 消息推送配置
+ * @return MqttPushSessionManager 实例
+ */
+ @Bean
+ public MqttPushSessionManager mqttPushSessionManager(MqttClientTemplate client,
+ MessageProperties messageProperties) {
+ return new MqttPushSessionManager(client, messageProperties);
+ }
+}
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
index f35a4a273..ed7354b1f 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
@@ -22,6 +22,16 @@ public interface MessageConstants {
*/
String MESSAGE_TOPIC = "global:message";
+ /**
+ * MQTT 广播消息主题后缀。
+ */
+ String MQTT_BROADCAST_TOPIC_SUFFIX = "/all";
+
+ /**
+ * MQTT 指定用户消息主题目录。
+ */
+ String MQTT_USER_TOPIC_PATH = "/user/";
+
/**
* 心跳请求标识
*/
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/MqttPushSessionManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/MqttPushSessionManager.java
new file mode 100644
index 000000000..70a5f755d
--- /dev/null
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/MqttPushSessionManager.java
@@ -0,0 +1,120 @@
+package org.dromara.common.push.core;
+
+import cn.hutool.core.collection.CollUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.json.utils.JsonUtils;
+import org.dromara.common.push.constant.MessageConstants;
+import org.dromara.common.push.dto.PushDTO;
+import org.dromara.common.push.properties.MessageProperties;
+import org.dromara.mica.mqtt.spring.client.MqttClientTemplate;
+import org.dromara.system.api.domain.PushPayloadDTO;
+
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+
+/**
+ * MQTT 推送会话管理器。
+ *
+ * @author Lion Li
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class MqttPushSessionManager implements PushSessionManager {
+
+ private final MqttClientTemplate client;
+
+ private final MessageProperties messageProperties;
+
+ /**
+ * MQTT 推送不依赖 Redis 主题订阅。
+ *
+ * @param consumer 消息消费逻辑
+ */
+ @Override
+ public void subscribeMessage(Consumer consumer) {
+ // MQTT 通道直接发布到 broker,不需要订阅 Redis 分发。
+ }
+
+ /**
+ * 发布指定用户 MQTT 消息。
+ *
+ * @param userId 目标用户ID
+ * @param payload 消息体
+ */
+ @Override
+ public void sendMessage(Long userId, PushPayloadDTO payload) {
+ if (userId == null || payload == null) {
+ return;
+ }
+ publish(topicPrefix() + MessageConstants.MQTT_USER_TOPIC_PATH + userId, payload);
+ }
+
+ /**
+ * 发布 MQTT 广播消息。
+ *
+ * @param payload 消息体
+ */
+ @Override
+ public void sendMessage(PushPayloadDTO payload) {
+ if (payload == null) {
+ return;
+ }
+ publish(topicPrefix() + MessageConstants.MQTT_BROADCAST_TOPIC_SUFFIX, payload);
+ }
+
+ /**
+ * 发布 MQTT 消息。
+ *
+ * @param pushDTO 推送参数封装对象
+ */
+ @Override
+ public void publishMessage(PushDTO pushDTO) {
+ if (pushDTO == null || pushDTO.getPayload() == null) {
+ return;
+ }
+ if (CollUtil.isEmpty(pushDTO.getUserIds())) {
+ sendMessage(pushDTO.getPayload());
+ return;
+ }
+ pushDTO.getUserIds().forEach(userId -> sendMessage(userId, pushDTO.getPayload()));
+ }
+
+ /**
+ * 发布 MQTT 广播消息。
+ *
+ * @param payload 消息体
+ */
+ @Override
+ public void publishAll(PushPayloadDTO payload) {
+ sendMessage(payload);
+ }
+
+ /**
+ * 发送消息到 MQTT Broker。
+ *
+ * @param topic MQTT 主题
+ * @param payload 消息体
+ */
+ private void publish(String topic, PushPayloadDTO payload) {
+ String message = JsonUtils.toJsonString(payload);
+ client.publish(topic, message.getBytes(StandardCharsets.UTF_8));
+ log.info("MQTT发送推送消息topic:{} message:{}", topic, payload.getMessage());
+ }
+
+ /**
+ * 获取规范化后的 MQTT topic 前缀。
+ *
+ * @return MQTT topic 前缀
+ */
+ private String topicPrefix() {
+ String prefix = messageProperties.getMqttTopicPrefix();
+ if (prefix == null || prefix.isBlank()) {
+ return "/message";
+ }
+ while (prefix.endsWith("/") && prefix.length() > 1) {
+ prefix = prefix.substring(0, prefix.length() - 1);
+ }
+ return prefix;
+ }
+}
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java
index 362d26645..0efc3d071 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java
@@ -24,7 +24,13 @@ public enum MessageTransportEnum {
* WebSocket 传输方式
* 全双工长连接,支持双向实时通信
*/
- WEBSOCKET("websocket");
+ WEBSOCKET("websocket"),
+
+ /**
+ * MQTT 传输方式
+ * 通过 MQTT Broker 发布统一推送消息
+ */
+ MQTT("mqtt");
/**
* 传输类型编码
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java
index 7d0aeaebf..e20ccd996 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java
@@ -19,7 +19,7 @@ public class MessageProperties {
private Boolean enabled = true;
/**
- * 传输方式:sse / websocket。
+ * 传输方式:sse / websocket / mqtt。
*/
private String transport = MessageTransportEnum.SSE.getCode();
@@ -52,4 +52,10 @@ public class MessageProperties {
* WebSocket 发送缓冲区大小。
*/
private int webSocketBufferSizeLimit = 64_000;
+
+ /**
+ * MQTT 消息推送主题前缀。
+ */
+ private String mqttTopicPrefix = "/message";
+
}
diff --git a/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
index 2e02f912d..db1bfd7a5 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+++ b/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
@@ -1,3 +1,4 @@
org.dromara.common.push.config.MessageAutoConfiguration
org.dromara.common.push.config.MessageSseConfiguration
org.dromara.common.push.config.MessageWebSocketConfiguration
+org.dromara.common.push.config.MessageMqttConfiguration