THENG
/
a-tuoheng-device
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

先注册再发送命令

This commit is contained in:
孙小云 2026-02-11 10:12:14 +08:00
parent 96561c749c
commit 8f85fc7a1e
1 changed files with 55 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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);
@ -179,27 +209,17 @@ public class TransactionExecutor {
return CompletableFuture.completedFuture(result);
}
// c. 等待方法回调（异步）
CallbackConfig methodCallback = instruction.getMethodCallbackConfig(context);
if (methodCallback != null) {
// 自动设置为方法回调类型
methodCallback.setCallbackType(CallbackConfig.CallbackType.METHOD);
return waitForCallbackAsync(methodCallback, context)
.thenCompose(methodResult -> {
// e. 等待方法回调（已经预注册）
if (finalMethodFuture != null) {
return finalMethodFuture.thenCompose(methodResult -> {
if (!methodResult.isSuccess()) {
instruction.onComplete(context, methodResult);
return CompletableFuture.completedFuture(methodResult);
}
// d. 等待状态回调（异步）
CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
if (stateCallback != null) {
// 自动设置为状态回调类型
stateCallback.setCallbackType(CallbackConfig.CallbackType.STATE);
return waitForCallbackAsync(stateCallback, context)
.thenApply(stateResult -> {
// f. 等待状态回调（已经预注册）
if (finalStateFuture != null) {
return finalStateFuture.thenApply(stateResult -> {
instruction.onComplete(context, stateResult);
return stateResult;
});
@ -213,13 +233,8 @@ public class TransactionExecutor {
}
// 没有方法回调，检查是否有状态回调
CallbackConfig stateCallback = instruction.getStateCallbackConfig(context);
if (stateCallback != null) {
// 自动设置为状态回调类型
stateCallback.setCallbackType(CallbackConfig.CallbackType.STATE);
return waitForCallbackAsync(stateCallback, context)
.thenApply(stateResult -> {
if (finalStateFuture != null) {
return finalStateFuture.thenApply(stateResult -> {
instruction.onComplete(context, stateResult);
return stateResult;
});