修改回调机制

src/main/java/com/ruoyi/device/domain/impl/machine/command/TransactionExecutor.java

@@ -162,7 +162,37 @@ public class TransactionExecutor {
             return CompletableFuture.completedFuture(result);
         }
 
-        // b. 在线程池中执行远程调用（避免阻塞当前线程）
+        // b. 预先获取回调配置（在发送命令前）
+        CallbackConfig methodCallback = instruction.getMethodCallbackConfig(context);
+        CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
+
+        // 设置回调类型
+        if (methodCallback != null) {
+            methodCallback.setCallbackType(CallbackConfig.CallbackType.METHOD);
+        }
+        if (stateCallback != null) {
+            stateCallback.setCallbackType(CallbackConfig.CallbackType.STATE);
+        }
+
+        // c. 预先注册回调（在发送命令前，避免竞态条件）
+        CompletableFuture<InstructionResult> methodFuture = null;
+        CompletableFuture<InstructionResult> stateFuture = null;
+
+        if (methodCallback != null) {
+            log.info("【预注册方法回调】instruction={}, topic={}",
+                instruction.getName(), methodCallback.getTopic());
+            methodFuture = waitForCallbackAsync(methodCallback, context);
+        }
+        if (stateCallback != null) {
+            log.info("【预注册状态回调】instruction={}, topic={}",
+                instruction.getName(), stateCallback.getTopic());
+            stateFuture = waitForCallbackAsync(stateCallback, context);
+        }
+
+        // d. 在线程池中执行远程调用（回调已经注册好了）
+        CompletableFuture<InstructionResult> finalMethodFuture = methodFuture;
+        CompletableFuture<InstructionResult> finalStateFuture = stateFuture;
+
         return CompletableFuture.supplyAsync(() -> {
             try {
                 instruction.executeRemoteCall(context);
@@ -174,55 +204,49 @@ public class TransactionExecutor {
             }
         }, commandExecutor).thenCompose(remoteCallSuccess -> {
             if (!remoteCallSuccess) {
+                // 命令发送失败，取消已注册的回调（避免资源泄漏）
+                if (finalMethodFuture != null) {
+                    log.warn("命令发送失败，取消方法回调");
+                    finalMethodFuture.cancel(true);
+                }
+                if (finalStateFuture != null) {
+                    log.warn("命令发送失败，取消状态回调");
+                    finalStateFuture.cancel(true);
+                }
                 InstructionResult result = InstructionResult.failure("远程调用失败");
                 instruction.onComplete(context, result);
                 return CompletableFuture.completedFuture(result);
             }
 
-            // c. 等待方法回调（异步）
-            CallbackConfig methodCallback = instruction.getMethodCallbackConfig(context);
-            if (methodCallback != null) {
-                // 自动设置为方法回调类型
-                methodCallback.setCallbackType(CallbackConfig.CallbackType.METHOD);
+            // e. 等待方法回调（已经预注册）
+            if (finalMethodFuture != null) {
+                return finalMethodFuture.thenCompose(methodResult -> {
+                    if (!methodResult.isSuccess()) {
+                        instruction.onComplete(context, methodResult);
+                        return CompletableFuture.completedFuture(methodResult);
+                    }
 
-                return waitForCallbackAsync(methodCallback, context)
-                    .thenCompose(methodResult -> {
-                        if (!methodResult.isSuccess()) {
-                            instruction.onComplete(context, methodResult);
-                            return CompletableFuture.completedFuture(methodResult);
-                        }
+                    // f. 等待状态回调（已经预注册）
+                    if (finalStateFuture != null) {
+                        return finalStateFuture.thenApply(stateResult -> {
+                            instruction.onComplete(context, stateResult);
+                            return stateResult;
+                        });
+                    }
 
-                        // d. 等待状态回调（异步）
-                        CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
-                        if (stateCallback != null) {
-                            // 自动设置为状态回调类型
-                            stateCallback.setCallbackType(CallbackConfig.CallbackType.STATE);
-
-                            return waitForCallbackAsync(stateCallback, context)
-                                .thenApply(stateResult -> {
-                                    instruction.onComplete(context, stateResult);
-                                    return stateResult;
-                                });
-                        }
-
-                        // 没有状态回调，直接成功
-                        InstructionResult result = InstructionResult.success();
-                        instruction.onComplete(context, result);
-                        return CompletableFuture.completedFuture(result);
-                    });
+                    // 没有状态回调，直接成功
+                    InstructionResult result = InstructionResult.success();
+                    instruction.onComplete(context, result);
+                    return CompletableFuture.completedFuture(result);
+                });
             }
 
             // 没有方法回调，检查是否有状态回调
-            CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
-            if (stateCallback != null) {
-                // 自动设置为状态回调类型
-                stateCallback.setCallbackType(CallbackConfig.CallbackType.STATE);
-
-                return waitForCallbackAsync(stateCallback, context)
-                    .thenApply(stateResult -> {
-                        instruction.onComplete(context, stateResult);
-                        return stateResult;
-                    });
+            if (finalStateFuture != null) {
+                return finalStateFuture.thenApply(stateResult -> {
+                    instruction.onComplete(context, stateResult);
+                    return stateResult;
+                });
             }
 
             // 没有任何回调，直接成功