【功能完善】IoT: 新增 MQTT RPC 支持，包含请求和响应模型、序列化工具、MQTT 配置及客户端/服务器实现，提供示例服务和控制器接口，优化插件结构以支持 HTTP 插件的集成。

yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/common/RpcRequest.java

@@ -0,0 +1,37 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.common;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * MQTT RPC 请求
+ *
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class RpcRequest {
+
+    /**
+     * 方法名
+     */
+    private String method;
+
+    /**
+     * 参数
+     */
+    private Object[] params;
+
+    /**
+     * 关联 ID
+     */
+    private String correlationId;
+
+    /**
+     * 回复地址
+     */
+    private String replyTo;
+}

yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/common/RpcResponse.java

@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.common;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * MQTT RPC 响应
+ *
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class RpcResponse {
+
+    /**
+     * 关联 ID
+     */
+    private String correlationId;
+
+    /**
+     * 结果
+     */
+    private Object result;
+
+    /**
+     * 错误
+     */
+    private String error;
+}

yudao-module-iot/yudao-module-iot-api/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/common/SerializationUtils.java

@@ -0,0 +1,18 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.common;
+
+import cn.hutool.json.JSONUtil;
+
+/**
+ * 序列化工具类
+ *
+ */
+public class SerializationUtils {
+
+    public static String serialize(Object obj) {
+        return JSONUtil.toJsonStr(obj);
+    }
+
+    public static <T> T deserialize(String json, Class<T> clazz) {
+        return JSONUtil.toBean(json, clazz);
+    }
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java

@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "mqtt")
+public class MqttConfig {
+    /**
+     * MQTT 代理地址    
+     */
+    private String broker;
+
+    /**
+     * MQTT 用户名
+     */
+    private String username;
+
+    /**
+     * MQTT 密码
+     */
+    private String password;
+
+    /**
+     * MQTT 客户端 ID
+     */
+    private String clientId;
+
+    /**
+     * MQTT 请求主题
+     */
+    private String requestTopic;
+
+    /**
+     * MQTT 响应主题前缀
+     */
+    private String responseTopicPrefix;
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/server/RpcServer.java

@@ -0,0 +1,95 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.server;
+
+import cn.hutool.core.lang.UUID;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils;
+import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.HashMap;
+import java.util.Map;
+
+@Service
+@Slf4j
+public class RpcServer {
+
+    private final MqttConfig mqttConfig;
+    private final MqttClient mqttClient;
+    private final Map<String, MethodInvoker> methodRegistry = new HashMap<>();
+
+    public RpcServer(MqttConfig mqttConfig) throws MqttException {
+        this.mqttConfig = mqttConfig;
+        this.mqttClient = new MqttClient(mqttConfig.getBroker(), "rpc-server-" + UUID.randomUUID(), new MemoryPersistence());
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setAutomaticReconnect(true);
+        options.setCleanSession(true);
+        options.setUserName(mqttConfig.getUsername());
+        options.setPassword(mqttConfig.getPassword().toCharArray());
+        this.mqttClient.connect(options);
+    }
+
+    @PostConstruct
+    public void init() throws MqttException {
+        mqttClient.subscribe(mqttConfig.getRequestTopic(), this::handleRequest);
+        log.info("RPC Server subscribed to topic: {}", mqttConfig.getRequestTopic());
+    }
+
+    private void handleRequest(String topic, MqttMessage message) {
+        RpcRequest request = SerializationUtils.deserialize(new String(message.getPayload()), RpcRequest.class);
+        RpcResponse response = new RpcResponse();
+        response.setCorrelationId(request.getCorrelationId());
+
+        try {
+            MethodInvoker invoker = methodRegistry.get(request.getMethod());
+            if (invoker == null) {
+                throw new NoSuchMethodException("Unknown method: " + request.getMethod());
+            }
+            Object result = invoker.invoke(request.getParams());
+            response.setResult(result);
+        } catch (Exception e) {
+            response.setError(e.getMessage());
+            log.error("Error processing RPC request: {}", e.getMessage(), e);
+        }
+
+        String replyPayload = SerializationUtils.serialize(response);
+        MqttMessage replyMessage = new MqttMessage(replyPayload.getBytes());
+        replyMessage.setQos(1);
+        try {
+            mqttClient.publish(request.getReplyTo(), replyMessage);
+            log.info("Published response to {}", request.getReplyTo());
+        } catch (MqttException e) {
+            log.error("Failed to publish response: {}", e.getMessage(), e);
+        }
+    }
+
+    /**
+     * 注册可调用的方法
+     *
+     * @param methodName 方法名称
+     * @param invoker    方法调用器
+     */
+    public void registerMethod(String methodName, MethodInvoker invoker) {
+        methodRegistry.put(methodName, invoker);
+        log.info("Registered method: {}", methodName);
+    }
+
+    @PreDestroy
+    public void cleanup() throws MqttException {
+        mqttClient.disconnect();
+        log.info("RPC Server disconnected");
+    }
+
+    /**
+     * 方法调用器接口
+     */
+    @FunctionalInterface
+    public interface MethodInvoker {
+        Object invoke(Object[] params) throws Exception;
+    }
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/ExampleService.java

@@ -0,0 +1,43 @@
+package cn.iocoder.yudao.module.iot.service.plugin;
+
+import cn.iocoder.yudao.module.iot.mqttrpc.server.RpcServer;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+
+@Service
+@RequiredArgsConstructor
+public class ExampleService {
+
+    private final RpcServer rpcServer;
+
+    @PostConstruct
+    public void registerMethods() {
+        rpcServer.registerMethod("add", params -> {
+            if (params.length != 2) {
+                throw new IllegalArgumentException("add方法需要两个参数");
+            }
+            int a = ((Number) params[0]).intValue();
+            int b = ((Number) params[1]).intValue();
+            return add(a, b);
+        });
+
+        rpcServer.registerMethod("concat", params -> {
+            if (params.length != 2) {
+                throw new IllegalArgumentException("concat方法需要两个参数");
+            }
+            String str1 = params[0].toString();
+            String str2 = params[1].toString();
+            return concat(str1, str2);
+        });
+    }
+
+    private int add(int a, int b) {
+        return a + b;
+    }
+
+    private String concat(String a, String b) {
+        return a + b;
+    }
+}

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/PluginInfoServiceImpl.java

@@ -18,7 +18,6 @@ import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.multipart.MultipartFile;
 
-import javax.annotation.PostConstruct;
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.*;

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/pom.xml

@@ -150,5 +150,10 @@
             <artifactId>netty-all</artifactId>
             <version>4.1.63.Final</version> <!-- 版本可根据需要调整 -->
         </dependency>
+        <!-- MQTT -->
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+        </dependency>
     </dependencies>
 </project>

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/HttpPluginSpringbootApplication.java

@@ -0,0 +1,11 @@
+package cn.iocoder.yudao.module.iot;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class HttpPluginSpringbootApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(HttpPluginSpringbootApplication.class, args);
+    }
+}

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/controller/RpcController.java

@@ -0,0 +1,31 @@
+
+package cn.iocoder.yudao.module.iot.controller;
+
+import cn.iocoder.yudao.module.iot.mqttrpc.client.RpcClient;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import java.util.concurrent.CompletableFuture;
+
+@RestController
+@RequestMapping("/rpc")
+@RequiredArgsConstructor
+public class RpcController {
+
+    @Resource
+    private RpcClient rpcClient;
+
+    @PostMapping("/add")
+    public CompletableFuture<Object> add(@RequestParam int a, @RequestParam int b) throws Exception {
+        return rpcClient.call("add", new Object[]{a, b}, 10);
+    }
+
+    @PostMapping("/concat")
+    public CompletableFuture<Object> concat(@RequestParam String str1, @RequestParam String str2) throws Exception {
+        return rpcClient.call("concat", new Object[]{str1, str2}, 10);
+    }
+}

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/client/RpcClient.java

@@ -0,0 +1,95 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.client;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcRequest;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.RpcResponse;
+import cn.iocoder.yudao.module.iot.mqttrpc.common.SerializationUtils;
+import cn.iocoder.yudao.module.iot.mqttrpc.config.MqttConfig;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.UUID;
+import java.util.concurrent.*;
+
+@Service
+@Slf4j
+public class RpcClient {
+
+    private final MqttConfig mqttConfig;
+    private final MqttClient mqttClient;
+    private final ConcurrentMap<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+
+    public RpcClient(MqttConfig mqttConfig) throws MqttException {
+        this.mqttConfig = mqttConfig;
+        this.mqttClient = new MqttClient(mqttConfig.getBroker(), mqttConfig.getClientId(), new MemoryPersistence());
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setAutomaticReconnect(true);
+        options.setCleanSession(true);
+        options.setUserName(mqttConfig.getUsername());
+        options.setPassword(mqttConfig.getPassword().toCharArray());
+        this.mqttClient.connect(options);
+    }
+
+    @PostConstruct
+    public void init() throws MqttException {
+        mqttClient.subscribe(mqttConfig.getResponseTopicPrefix() + "#", this::handleResponse);
+        log.info("RPC Client subscribed to topics: {}", mqttConfig.getResponseTopicPrefix() + "#");
+    }
+
+    private void handleResponse(String topic, MqttMessage message) {
+        String correlationId = topic.substring(mqttConfig.getResponseTopicPrefix().length());
+        RpcResponse response = SerializationUtils.deserialize(new String(message.getPayload()), RpcResponse.class);
+        CompletableFuture<RpcResponse> future = pendingRequests.remove(correlationId);
+        if (future != null) {
+            if (response.getError() != null) {
+                future.completeExceptionally(new RuntimeException(response.getError()));
+            } else {
+                future.complete(response);
+            }
+        } else {
+            log.warn("Received response for unknown correlationId: {}", correlationId);
+        }
+    }
+
+    public CompletableFuture<Object> call(String method, Object[] params, int timeoutSeconds) throws MqttException {
+        String correlationId = UUID.randomUUID().toString();
+        String replyTo = mqttConfig.getResponseTopicPrefix() + correlationId;
+
+        RpcRequest request = new RpcRequest(method, params, correlationId, replyTo);
+        String payload = SerializationUtils.serialize(request);
+        MqttMessage message = new MqttMessage(payload.getBytes());
+        message.setQos(1);
+        mqttClient.publish(mqttConfig.getRequestTopic(), message);
+
+        CompletableFuture<RpcResponse> futureResponse = new CompletableFuture<>();
+        pendingRequests.put(correlationId, futureResponse);
+
+        // 设置超时
+        scheduler.schedule(() -> {
+            CompletableFuture<RpcResponse> removed = pendingRequests.remove(correlationId);
+            if (removed != null) {
+                removed.completeExceptionally(new TimeoutException("RPC call timed out"));
+            }
+        }, timeoutSeconds, TimeUnit.SECONDS);
+
+        // 返回最终的结果
+        return futureResponse.thenApply(RpcResponse::getResult);
+    }
+
+    @PreDestroy
+    public void cleanup() throws MqttException {
+        mqttClient.disconnect();
+        scheduler.shutdown();
+        log.info("RPC Client disconnected");
+    }
+}

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/java/cn/iocoder/yudao/module/iot/mqttrpc/config/MqttConfig.java

@@ -0,0 +1,41 @@
+package cn.iocoder.yudao.module.iot.mqttrpc.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "mqtt")
+public class MqttConfig {
+
+    /**
+     * MQTT 代理地址    
+     */
+    private String broker;
+
+    /**
+     * MQTT 用户名
+     */
+    private String username;
+
+    /**
+     * MQTT 密码
+     */
+    private String password;
+
+    /**
+     * MQTT 客户端 ID
+     */
+    private String clientId;
+
+    /**
+     * MQTT 请求主题
+     */
+    private String requestTopic;
+
+    /**
+     * MQTT 响应主题前缀
+     */
+    private String responseTopicPrefix;
+}

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-http-plugin/src/main/resources/application.yml

@@ -0,0 +1,15 @@
+server:
+  port: 8092
+
+spring:
+  application:
+    name: yudao-module-iot-http-plugin
+  
+# MQTT-RPC 配置
+mqtt:
+  broker: tcp://chaojiniu.top:1883
+  username: haohao
+  password: ahh@123456
+  clientId: mqtt-rpc-client-${random.int}
+  requestTopic: rpc/request
+  responseTopicPrefix: rpc/response/

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-mqtt-plugin/pom.xml

@@ -145,10 +145,10 @@
             <version>${lombok.version}</version>
             <scope>provided</scope>
         </dependency>
+        <!-- MQTT -->
         <dependency>
-            <groupId>io.netty</groupId>
+            <groupId>org.eclipse.paho</groupId>
-            <artifactId>netty-all</artifactId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
-            <version>4.1.63.Final</version> <!-- 版本可根据需要调整 -->
         </dependency>
     </dependencies>
 </project>

yudao-module-iot/yudao-module-iot-plugin/yudao-module-iot-mqtt-plugin/src/main/java/cn/iocoder/yudao/module/iot/plugin/MqttPlugin.java

@@ -1,11 +1,12 @@
 package cn.iocoder.yudao.module.iot.plugin;
 
-import cn.iocoder.yudao.module.iot.api.device.DeviceDataApi;
 import cn.iocoder.yudao.module.iot.api.ServiceRegistry;
+import cn.iocoder.yudao.module.iot.api.device.DeviceDataApi;
 import lombok.extern.slf4j.Slf4j;
-import org.pf4j.PluginWrapper;
 import org.pf4j.Plugin;
+import org.pf4j.PluginWrapper;
 
+import javax.annotation.Resource;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -13,11 +14,11 @@ import java.util.concurrent.Executors;
 public class MqttPlugin extends Plugin {
 
     private ExecutorService executorService;
+    @Resource
     private DeviceDataApi deviceDataApi;
 
     public MqttPlugin(PluginWrapper wrapper) {
         super(wrapper);
-        // 初始化线程池
         this.executorService = Executors.newSingleThreadExecutor();
     }
 
@@ -25,24 +26,20 @@ public class MqttPlugin extends Plugin {
     public void start() {
         log.info("MqttPlugin.start()");
 
-        // 重新初始化线程池，确保它是活跃的
         if (executorService.isShutdown() || executorService.isTerminated()) {
             executorService = Executors.newSingleThreadExecutor();
         }
 
-        // 从 ServiceRegistry 中获取主程序暴露的 DeviceDataApi 接口实例
         deviceDataApi = ServiceRegistry.getService(DeviceDataApi.class);
         if (deviceDataApi == null) {
             log.error("未能从 ServiceRegistry 获取 DeviceDataApi 实例，请确保主程序已正确注册！");
             return;
         }
+
     }
 
     @Override
     public void stop() {
         log.info("MqttPlugin.stop()");
-        // 停止线程池
-        executorService.shutdownNow();
     }
-
 }