+ * 用法文档 ...
+ *
+ * @author Lion Li
+ */
+@AutoConfiguration
+public class MqttAutoConfiguration {
+
+ @Bean
+ public MqttClientConnectListener mqttClientConnectListener(MqttClientCreator mqttClientCreator) {
+ return new MqttClientConnectListener(mqttClientCreator);
+ }
+
+ @Bean
+ public MqttClientGlobalMessageListener mqttClientGlobalMessageListener() {
+ return new MqttClientGlobalMessageListener();
+ }
+
+}
diff --git a/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientConnectListener.java b/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientConnectListener.java
new file mode 100644
index 000000000..fd53eeabd
--- /dev/null
+++ b/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientConnectListener.java
@@ -0,0 +1,37 @@
+package org.dromara.common.mqtt.listener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
+import org.dromara.mica.mqtt.core.client.MqttClientCreator;
+import org.tio.core.ChannelContext;
+
+/**
+ * 客户端连接状态监听
+ *
+ * @author Lion Li
+ */
+@Slf4j
+public class MqttClientConnectListener implements IMqttClientConnectListener {
+ //
+ private final MqttClientCreator mqttClientCreator;
+
+ public MqttClientConnectListener(MqttClientCreator mqttClientCreator) {
+ this.mqttClientCreator = mqttClientCreator;
+ }
+
+ @Override
+ public void onConnected(ChannelContext context, boolean isReconnect) {
+ // 创建连接
+ log.info("MqttConnectedEvent:{}", context);
+ }
+
+ @Override
+ public void onDisconnect(ChannelContext context, Throwable throwable, String remark, boolean isRemove) {
+ // 离线时更新重连
+ log.info("MqttDisconnectEvent:{}", context, throwable);
+ // 在断线时更新 clientId、username、password
+// mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
+// .username("newUserName")
+// .password("newPassword");
+ }
+}
diff --git a/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientGlobalMessageListener.java b/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientGlobalMessageListener.java
new file mode 100644
index 000000000..a3117eb74
--- /dev/null
+++ b/ruoyi-common/ruoyi-common-mqtt/src/main/java/org/dromara/common/mqtt/listener/MqttClientGlobalMessageListener.java
@@ -0,0 +1,23 @@
+package org.dromara.common.mqtt.listener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
+import org.dromara.mica.mqtt.core.client.IMqttClientGlobalMessageListener;
+import org.tio.core.ChannelContext;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * 全局消息监听，可以监听到所有订阅消息
+ *
+ * @author Lion Li
+ */
+@Slf4j
+public class MqttClientGlobalMessageListener implements IMqttClientGlobalMessageListener {
+
+ @Override
+ public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
+ log.info("MqttGlobalMessageEvent => topic: {}, msg: {}", topic, new String(payload, StandardCharsets.UTF_8));
+ }
+
+}
diff --git a/ruoyi-common/ruoyi-common-mqtt/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/ruoyi-common/ruoyi-common-mqtt/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
new file mode 100644
index 000000000..265d0ddcc
--- /dev/null
+++ b/ruoyi-common/ruoyi-common-mqtt/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
@@ -0,0 +1 @@
+org.dromara.common.mqtt.config.MqttAutoConfiguration
diff --git a/ruoyi-modules/ruoyi-demo/pom.xml b/ruoyi-modules/ruoyi-demo/pom.xml
index 0a91fdc62..61d7380ea 100644
--- a/ruoyi-modules/ruoyi-demo/pom.xml
+++ b/ruoyi-modules/ruoyi-demo/pom.xml
@@ -98,6 +98,11 @@
+ * 用法文档 ... + * + * @author Lion Li + */ +@RequiredArgsConstructor +@RestController +@RequestMapping("/demo/mqtt") +@Slf4j +public class MqttController { + + private final MqttClientTemplate client; + + @GetMapping("/send") + public boolean send() { + client.publish("/test/client", "测试测试".getBytes(StandardCharsets.UTF_8)); + return true; + } + + @MqttClientSubscribe("/test/#") + public void subQos0(String topic, byte[] payload) { + log.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8)); + } + + @MqttClientSubscribe(value = "/qos1/#", qos = MqttQoS.QOS1) + public void subQos1(String topic, byte[] payload) { + log.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8)); + } + + @MqttClientSubscribe("/sys/${productKey}/${deviceName}/thing/sub/register") + public void thingSubRegister(String topic, byte[] payload) { + // 1.3.8 开始支持，@MqttClientSubscribe 注解支持 ${} 变量替换，会默认替换成 + + // 注意：mica-mqtt 会先从 Spring boot 配置中替换参数 ${}，如果存在配置会优先被替换。 + log.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8)); + } + + @MqttClientSubscribe( + value = "/test/json", + deserialize = MqttJsonDeserializer.class // 2.4.5 开始支持 自定义序列化，默认 json 序列化 + ) + public void testJson(String topic, MqttPublishMessage message, TestDemo data) { + // 2.4.5 开始支持，支持 2 到 3 个参数，字段类型映射规则如下 + // String 字符串会默认映射到 topic， + // MqttPublishMessage 会默认映射到 原始的消息，可以拿到 mqtt5 的 props 参数 + // byte[] 会映射到 mqtt 消息内容 payload + // ByteBuffer 会映射到 mqtt 消息内容 payload + // 其他类型会走序列化，确保消息能够序列化，默认为 json 序列化 + log.info("topic:{} json data:{}", topic, data); + } + +}