迁移状态机

This commit is contained in:
孙小云 2026-01-28 11:27:02 +08:00
parent bef017bf63
commit a205823773
55 changed files with 4492 additions and 0 deletions

View File

@ -0,0 +1,272 @@
package com.ruoyi.device.domain.impl.machine;
import com.ruoyi.device.domain.impl.machine.command.*;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import com.ruoyi.device.domain.impl.machine.mqtt.MqttClient;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
import com.ruoyi.device.domain.impl.machine.statemachine.MachineStateManager;
import com.ruoyi.device.domain.impl.machine.statemachine.StateChangeListener;
import com.ruoyi.device.domain.impl.machine.vendor.VendorConfig;
import com.ruoyi.device.domain.impl.machine.vendor.VendorRegistry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
/**
* 设备命令管理器框架使用者的主要入口
*/
@Slf4j
@Component
public class MachineCommandManager {
private final VendorRegistry vendorRegistry;
private final MachineStateManager stateManager;
private final TransactionExecutor transactionExecutor;
private final MqttClient mqttClient;
/**
* SN -> 当前正在执行的命令
*/
private final Map<String, CommandExecution> executingCommands = new ConcurrentHashMap<>();
/**
* 命令执行监听器
*/
private final Map<String, CommandExecutionListener> commandListeners = new ConcurrentHashMap<>();
public MachineCommandManager(VendorRegistry vendorRegistry,
MachineStateManager stateManager,
TransactionExecutor transactionExecutor,
MqttClient mqttClient) {
this.vendorRegistry = vendorRegistry;
this.stateManager = stateManager;
this.transactionExecutor = transactionExecutor;
this.mqttClient = mqttClient;
}
/**
* 获取设备当前状态
*
* @param sn 设备SN号
* @return 设备状态
*/
public MachineStates getMachineStates(String sn) {
return stateManager.getStates(sn);
}
/**
* 更新设备状态通常在心跳中调用
*
* @param sn 设备SN号
* @param newStates 新状态
*/
public void updateMachineStates(String sn, MachineStates newStates,Boolean force) {
stateManager.updateStates(sn, newStates,force);
}
/**
* 判断设备是否正在执行命令
*
* @param sn 设备SN号
* @return 是否正在执行命令
*/
public boolean isExecutingCommand(String sn) {
CommandExecution execution = executingCommands.get(sn);
return execution != null && !execution.getFuture().isDone();
}
/**
* 获取设备当前正在执行的命令类型
*
* @param sn 设备SN号
* @return 命令类型如果没有正在执行的命令则返回null
*/
public CommandType getExecutingCommandType(String sn) {
CommandExecution execution = executingCommands.get(sn);
if (execution != null && !execution.getFuture().isDone()) {
return execution.getCommandType();
}
return null;
}
/**
* 获取设备在当前状态下可以执行的命令列表
*
* @param sn 设备SN号
* @return 可执行的命令列表
*/
public List<CommandType> getAvailableCommands(String sn) {
VendorConfig vendorConfig = vendorRegistry.getVendorConfig(sn);
if (vendorConfig == null) {
log.warn("设备未绑定厂家: sn={}", sn);
return List.of();
}
MachineStates currentStates = stateManager.getStates(sn);
return vendorConfig.getAvailableCommands(currentStates);
}
/**
* 执行命令
*
* @param sn 设备SN号
* @param commandType 命令类型
* @return 命令执行结果的Future
*/
public CompletableFuture<CommandResult> executeCommand(String sn, CommandType commandType) {
return executeCommand(sn, commandType, Map.of());
}
/**
* 执行命令带参数
*
* @param sn 设备SN号
* @param commandType 命令类型
* @param params 命令参数
* @return 命令执行结果的Future
*/
public CompletableFuture<CommandResult> executeCommand(String sn, CommandType commandType, Map<String, Object> params) {
log.info("收到命令执行请求: sn={}, commandType={}, params={}", sn, commandType, params);
// 1. 检查设备是否已绑定厂家
VendorConfig vendorConfig = vendorRegistry.getVendorConfig(sn);
if (vendorConfig == null) {
String error = "设备未绑定厂家";
log.error("{}: sn={}", error, sn);
return CompletableFuture.completedFuture(CommandResult.failure(commandType, error));
}
// 2. 检查是否正在执行其他命令
if (isExecutingCommand(sn)) {
String error = "设备正在执行其他命令: " + getExecutingCommandType(sn);
log.warn("{}: sn={}", error, sn);
return CompletableFuture.completedFuture(CommandResult.failure(commandType, error));
}
// 3. 检查当前状态是否可以执行该命令
MachineStates currentStates = stateManager.getStates(sn);
if (!vendorConfig.canExecuteCommand(currentStates, commandType)) {
String error = "当前状态不允许执行该命令";
log.warn("{}: sn={}, commandType={}, currentStates={}", error, sn, commandType, currentStates);
return CompletableFuture.completedFuture(CommandResult.failure(commandType, error));
}
// 4. 获取事务定义
Transaction transaction = vendorConfig.getTransaction(commandType);
if (transaction == null) {
String error = "厂家不支持该命令";
log.error("{}: sn={}, commandType={}, vendorType={}", error, sn, commandType, vendorConfig.getVendorType());
return CompletableFuture.completedFuture(CommandResult.failure(commandType, error));
}
// 5. 创建指令上下文
InstructionContext context = new InstructionContext(sn, vendorConfig.getVendorType(), mqttClient);
params.forEach(context::putCommandParam);
// 6. 执行事务
CompletableFuture<CommandResult> future = transactionExecutor.executeTransaction(transaction, context);
// 7. 记录正在执行的命令
executingCommands.put(sn, new CommandExecution(commandType, future, System.currentTimeMillis()));
// 8. 添加完成回调
future.whenComplete((result, throwable) -> {
executingCommands.remove(sn);
if (throwable != null) {
log.error("命令执行异常: sn={}, commandType={}", sn, commandType, throwable);
notifyCommandComplete(sn, CommandResult.failure(commandType, "命令执行异常: " + throwable.getMessage()));
} else {
log.info("命令执行完成: sn={}, commandType={}, success={}", sn, commandType, result.isSuccess());
notifyCommandComplete(sn, result);
}
});
return future;
}
/**
* 注册命令执行监听器
*
* @param listenerId 监听器ID
* @param listener 监听器
*/
public void registerCommandListener(String listenerId, CommandExecutionListener listener) {
commandListeners.put(listenerId, listener);
log.debug("注册命令执行监听器: listenerId={}", listenerId);
}
/**
* 取消注册命令执行监听器
*
* @param listenerId 监听器ID
*/
public void unregisterCommandListener(String listenerId) {
commandListeners.remove(listenerId);
log.debug("取消注册命令执行监听器: listenerId={}", listenerId);
}
/**
* 注册状态变化监听器
*
* @param listenerId 监听器ID
* @param listener 监听器
*/
public void registerStateChangeListener(String listenerId, StateChangeListener listener) {
stateManager.registerStateChangeListener(listenerId, listener);
}
/**
* 取消注册状态变化监听器
*
* @param listenerId 监听器ID
*/
public void unregisterStateChangeListener(String listenerId) {
stateManager.unregisterStateChangeListener(listenerId);
}
/**
* 通知命令执行完成
*/
private void notifyCommandComplete(String sn, CommandResult result) {
for (CommandExecutionListener listener : commandListeners.values()) {
try {
listener.onCommandComplete(sn, result);
} catch (Exception e) {
log.error("命令执行监听器执行失败: sn={}, commandType={}", sn, result.getCommandType(), e);
}
}
}
/**
* 命令执行信息
*/
private static class CommandExecution {
private final CommandType commandType;
private final CompletableFuture<CommandResult> future;
private final long startTime;
public CommandExecution(CommandType commandType, CompletableFuture<CommandResult> future, long startTime) {
this.commandType = commandType;
this.future = future;
this.startTime = startTime;
}
public CommandType getCommandType() {
return commandType;
}
public CompletableFuture<CommandResult> getFuture() {
return future;
}
public long getStartTime() {
return startTime;
}
}
}

View File

@ -0,0 +1,15 @@
package com.ruoyi.device.domain.impl.machine.command;
/**
* 命令执行监听器
*/
@FunctionalInterface
public interface CommandExecutionListener {
/**
* 命令执行完成回调
*
* @param sn 设备SN号
* @param result 命令执行结果
*/
void onCommandComplete(String sn, CommandResult result);
}

View File

@ -0,0 +1,58 @@
package com.ruoyi.device.domain.impl.machine.command;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 命令执行结果
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class CommandResult {
/**
* 命令类型
*/
private CommandType commandType;
/**
* 是否成功
*/
private boolean success;
/**
* 错误信息
*/
private String errorMessage;
/**
* 失败的指令名称
*/
private String failedInstructionName;
/**
* 结果数据
*/
private Object data;
public static CommandResult success(CommandType commandType) {
return new CommandResult(commandType, true, null, null, null);
}
public static CommandResult success(CommandType commandType, Object data) {
return new CommandResult(commandType, true, null, null, data);
}
public static CommandResult failure(CommandType commandType, String errorMessage) {
return new CommandResult(commandType, false, errorMessage, null, null);
}
public static CommandResult failure(CommandType commandType, String errorMessage, String failedInstructionName) {
return new CommandResult(commandType, false, errorMessage, failedInstructionName, null);
}
public static CommandResult timeout(CommandType commandType) {
return new CommandResult(commandType, false, "命令执行超时", null, null);
}
}

View File

@ -0,0 +1,86 @@
package com.ruoyi.device.domain.impl.machine.command;
/**
* 命令类型枚举
*/
public enum CommandType {
/**
* 起飞
*/
TAKE_OFF,
/**
* 返航
*/
RETURN_HOME,
/**
* 急停
*/
EMERGENCY_STOP,
/**
* 继续飞行
*/
RESUME_FLIGHT,
/**
* 指点飞行
*/
POINT_FLY,
/**
* 取消指点
*/
CANCEL_POINT,
/**
* 开始航线任务
*/
START_MISSION,
/**
* 暂停航线任务
*/
PAUSE_MISSION,
/**
* 恢复航线任务
*/
RESUME_MISSION,
/**
* 打开舱门
*/
OPEN_COVER,
/**
* 关闭舱门
*/
CLOSE_COVER,
/**
* 进入调试模式
*/
ENTER_DEBUG_MODE,
/**
* 退出调试模式
*/
EXIT_DEBUG_MODE,
/**
* 进入DRC模式
*/
ENTER_DRC_MODE,
/**
* 退出DRC模式
*/
EXIT_DRC_MODE,
/**
* 重启机巢
*/
REBOOT_AIRPORT
}

View File

@ -0,0 +1,62 @@
package com.ruoyi.device.domain.impl.machine.command;
import com.ruoyi.device.domain.impl.machine.instruction.Instruction;
import lombok.Data;
/**
* 事务由多个指令组成的树状结构支持条件分支
*/
@Data
public class Transaction {
/**
* 事务名称
*/
private String name;
/**
* 命令类型
*/
private CommandType commandType;
/**
* 根指令事务的起始指令
*/
private Instruction rootInstruction;
/**
* 事务超时时间毫秒
*/
private long timeoutMs = 20000; // 默认10秒
public Transaction(String name, CommandType commandType) {
this.name = name;
this.commandType = commandType;
}
/**
* 设置根指令
*
* @param instruction 根指令
* @return Transaction 支持链式调用
*/
public Transaction root(Instruction instruction) {
this.rootInstruction = instruction;
return this;
}
/**
* 获取根指令
*/
public Instruction getRootInstruction() {
return rootInstruction;
}
/**
* 设置超时时间
*/
public Transaction setTimeout(long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
}

View File

@ -0,0 +1,313 @@
package com.ruoyi.device.domain.impl.machine.command;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.Instruction;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionResult;
import com.ruoyi.device.domain.impl.machine.mqtt.MqttCallbackRegistry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 事务执行器完全异步化版本
*
* 设计说明
* 1. 完全异步不阻塞任何线程所有操作都通过 CompletableFuture 链式调用
* 2. 高并发可以同时处理数万个命令不会创建大量线程
* 3. 资源高效线程只在真正需要执行任务时才使用不会浪费在等待上
*
* 性能优势
* - 传统方式10万个命令 = 10万个阻塞线程 = 系统崩溃
* - 异步方式10万个命令 = 200个工作线程 + 10万个 CompletableFuture = 正常运行
*/
@Slf4j
@Component
public class TransactionExecutor {
private final MqttCallbackRegistry callbackRegistry;
private final Executor commandExecutor;
private final ScheduledExecutorService timeoutScheduler;
public TransactionExecutor(
MqttCallbackRegistry callbackRegistry,
@Qualifier("commandExecutor") Executor commandExecutor) {
this.callbackRegistry = callbackRegistry;
this.commandExecutor = commandExecutor;
// 创建一个专门用于超时检查的调度器核心线程数较小
this.timeoutScheduler = new ScheduledThreadPoolExecutor(
2,
r -> {
Thread t = new Thread(r, "timeout-scheduler");
t.setDaemon(true);
return t;
}
);
log.info("事务执行器初始化完成(完全异步模式)");
}
/**
* 执行事务完全异步
*
* @param transaction 事务定义
* @param context 执行上下文
* @return CompletableFuture不会阻塞调用线程
*/
public CompletableFuture<CommandResult> executeTransaction(Transaction transaction, InstructionContext context) {
log.info("开始执行事务: transaction={}, sn={}", transaction.getName(), context.getSn());
long startTime = System.currentTimeMillis();
// 直接返回异步执行的结果不创建新线程
return executeInstructionTreeAsync(transaction, context, startTime, transaction.getRootInstruction());
}
/**
* 异步执行指令树
*
* @param transaction 事务定义
* @param context 执行上下文
* @param startTime 事务开始时间
* @param currentInstruction 当前要执行的指令
* @return CompletableFuture
*/
private CompletableFuture<CommandResult> executeInstructionTreeAsync(
Transaction transaction,
InstructionContext context,
long startTime,
Instruction currentInstruction) {
// 检查根指令
if (currentInstruction == null) {
log.error("事务没有根指令: transaction={}", transaction.getName());
return CompletableFuture.completedFuture(
CommandResult.failure(transaction.getCommandType(), "事务没有根指令")
);
}
// 检查事务是否超时
if (System.currentTimeMillis() - startTime > transaction.getTimeoutMs()) {
log.warn("事务执行超时: transaction={}, sn={}", transaction.getName(), context.getSn());
return CompletableFuture.completedFuture(
CommandResult.timeout(transaction.getCommandType())
);
}
log.debug("执行指令: instruction={}", currentInstruction.getName());
// 异步执行当前指令
Instruction finalCurrentInstruction = currentInstruction;
return executeInstructionAsync(currentInstruction, context)
.thenCompose(result -> {
// 根据执行结果获取下游指令
Instruction nextInstruction = finalCurrentInstruction.getNextInstruction(result.isSuccess());
if (nextInstruction != null) {
// 有下游指令递归执行
log.debug("根据执行结果选择下游指令: success={}, nextInstruction={}",
result.isSuccess(), nextInstruction.getName());
return executeInstructionTreeAsync(transaction, context, startTime, nextInstruction);
} else {
// 没有下游指令当前指令的结果就是事务的结果
if (!result.isSuccess()) {
log.error("指令执行失败(无下游指令): instruction={}, error={}",
finalCurrentInstruction.getName(), result.getErrorMessage());
return CompletableFuture.completedFuture(
CommandResult.failure(
transaction.getCommandType(),
result.getErrorMessage(),
finalCurrentInstruction.getName()
)
);
} else {
log.info("指令执行成功(无下游指令),事务完成: instruction={}, sn={}",
finalCurrentInstruction.getName(), context.getSn());
return CompletableFuture.completedFuture(
CommandResult.success(transaction.getCommandType())
);
}
}
});
}
/**
* 异步执行单个指令
*
* @param instruction 指令
* @param context 执行上下文
* @return CompletableFuture<InstructionResult>
*/
private CompletableFuture<InstructionResult> executeInstructionAsync(
Instruction instruction,
InstructionContext context) {
log.debug("开始执行指令: instruction={}, sn={}", instruction.getName(), context.getSn());
// a. 判断是否可以执行
if (!instruction.canExecute(context)) {
String error = "指令被拒绝";
log.warn("指令不满足执行条件: instruction={}, sn={}", instruction.getName(), context.getSn());
InstructionResult result = InstructionResult.failure(error);
instruction.onComplete(context, result);
return CompletableFuture.completedFuture(result);
}
// b. 在线程池中执行远程调用避免阻塞当前线程
return CompletableFuture.supplyAsync(() -> {
try {
instruction.executeRemoteCall(context);
log.debug("远程调用已发送: instruction={}", instruction.getName());
return true;
} catch (Exception e) {
log.error("远程调用失败: instruction={}, sn={}", instruction.getName(), context.getSn(), e);
return false;
}
}, commandExecutor).thenCompose(remoteCallSuccess -> {
if (!remoteCallSuccess) {
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);
return waitForCallbackAsync(methodCallback, context)
.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 -> {
instruction.onComplete(context, stateResult);
return stateResult;
});
}
// 没有状态回调直接成功
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;
});
}
// 没有任何回调直接成功
InstructionResult result = InstructionResult.success();
instruction.onComplete(context, result);
return CompletableFuture.completedFuture(result);
});
}
/**
* 异步等待回调不阻塞线程
*
* 关键改进
* 1. 不使用 future.get() 阻塞线程
* 2. 使用 ScheduledExecutorService 实现超时
* 3. 完全基于回调机制
*
* @param callbackConfig 回调配置
* @param context 执行上下文
* @return CompletableFuture<InstructionResult>
*/
private CompletableFuture<InstructionResult> waitForCallbackAsync(
CallbackConfig callbackConfig,
InstructionContext context) {
CompletableFuture<InstructionResult> future = new CompletableFuture<>();
AtomicBoolean callbackReceived = new AtomicBoolean(false);
// 注册回调包含 tid/bid 过滤
String callbackId = callbackRegistry.registerCallback(
callbackConfig.getTopic(),
messageBody -> {
// 判断消息是否匹配
boolean matches = callbackConfig.matches(messageBody);
if (matches) {
// 匹配成功
if (callbackReceived.compareAndSet(false, true)) {
future.complete(InstructionResult.success(messageBody));
log.debug("收到匹配的回调消息: topic={}, type={}",
callbackConfig.getTopic(), callbackConfig.getCallbackType());
}
} else {
// 不匹配根据回调类型决定行为
if (callbackConfig.getCallbackType() == CallbackConfig.CallbackType.METHOD) {
// 方法回调不匹配就失败
if (callbackReceived.compareAndSet(false, true)) {
future.complete(InstructionResult.failure("方法回调不匹配"));
log.warn("方法回调不匹配,指令失败: topic={}, expected={}, actual={}",
callbackConfig.getTopic(),
callbackConfig.getExpectedValue(),
messageBody);
}
} else {
// 状态回调不匹配继续等待
// 使用 CAS 确保只处理一次然后重置状态
if (callbackReceived.compareAndSet(false, true)) {
callbackReceived.set(false); // 重置状态继续等待下一条消息
log.debug("状态回调不匹配,继续等待: topic={}, expected={}, actual={}",
callbackConfig.getTopic(),
callbackConfig.getExpectedValue(),
messageBody);
}
}
}
},
callbackConfig.getTimeoutMs(),
callbackConfig.getTidFieldPath(),
callbackConfig.getExpectedTid(),
callbackConfig.getBidFieldPath(),
callbackConfig.getExpectedBid()
);
// 设置超时不阻塞线程
timeoutScheduler.schedule(() -> {
// 使用 CAS 确保只处理一次
if (callbackReceived.compareAndSet(false, true)) {
future.complete(InstructionResult.timeout());
log.warn("等待回调超时: topic={}, timeout={}ms",
callbackConfig.getTopic(), callbackConfig.getTimeoutMs());
}
}, callbackConfig.getTimeoutMs(), TimeUnit.MILLISECONDS);
// 清理回调无论成功还是超时
return future.whenComplete((result, throwable) -> {
callbackRegistry.unregisterCallback(callbackId);
});
}
}

View File

@ -0,0 +1,72 @@
package com.ruoyi.device.domain.impl.machine.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
/**
* 线程池配置
* 用于命令执行的异步任务
*/
@Slf4j
@Configuration
public class ExecutorConfig {
/**
* 命令执行线程池
*
* 设计说明
* 1. 核心线程数CPU 核心数 * 2适合 I/O 密集型任务
* 2. 最大线程数200控制最大并发防止线程爆炸
* 3. 队列容量10000缓冲等待执行的任务
* 4. 拒绝策略CallerRunsPolicy背压机制让调用者线程执行
*
* 性能预估
* - 假设每个命令平均执行 10
* - 200 个线程可以同时处理 200 个命令
* - 队列可以缓冲 10000 个命令
* - 总容量10200 个并发命令
* - 吞吐量200 / 10 = 20 个命令/
*/
@Bean(name = "commandExecutor")
public Executor commandExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 核心线程数根据 CPU 核心数设置
int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
executor.setCorePoolSize(corePoolSize);
log.info("命令执行线程池核心线程数: {}", corePoolSize);
// 最大线程数限制最大并发防止线程爆炸
executor.setMaxPoolSize(200);
// 队列容量缓冲等待执行的任务
executor.setQueueCapacity(10000);
// 拒绝策略队列满时调用者线程执行背压机制
// 这样可以防止任务丢失同时给系统施加背压
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 线程名称前缀方便日志追踪
executor.setThreadNamePrefix("cmd-exec-");
// 等待任务完成后再关闭优雅关闭
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(60);
// 允许核心线程超时节省资源
executor.setAllowCoreThreadTimeOut(true);
executor.setKeepAliveSeconds(60);
executor.initialize();
log.info("命令执行线程池初始化完成: corePoolSize={}, maxPoolSize={}, queueCapacity={}",
corePoolSize, 200, 10000);
return executor;
}
}

View File

@ -0,0 +1,29 @@
package com.ruoyi.device.domain.impl.machine.config;
import com.ruoyi.device.domain.impl.machine.vendor.VendorRegistry;
import com.ruoyi.device.domain.impl.machine.vendor.dji.DjiVendorConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 设备框架配置类
*/
@Slf4j
@Configuration
public class MachineFrameworkConfig {
/**
* 自动注册所有厂家配置
*/
@Bean
public CommandLineRunner registerVendors(VendorRegistry vendorRegistry, DjiVendorConfig djiVendorConfig) {
return args -> {
// 注册大疆厂家配置
vendorRegistry.registerVendor(djiVendorConfig);
log.info("设备框架初始化完成,已注册厂家: {}", vendorRegistry.getAllVendorTypes());
};
}
}

View File

@ -0,0 +1,30 @@
package com.ruoyi.device.domain.impl.machine.config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
/**
* Redis 配置类
* 用于配置 Redis Pub/Sub 相关的 Bean
*/
@Configuration
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "redis")
public class RedisConfig {
/**
* 创建 Redis 消息监听容器专用于机器框架的 MQTT 回调
* 用于 Redis Pub/Sub 功能
*
* 注意使用特定的 Bean 名称避免与其他模块冲突
*/
@Bean(name = "machineFrameworkRedisMessageListenerContainer")
public RedisMessageListenerContainer machineFrameworkRedisMessageListenerContainer(
RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
return container;
}
}

View File

@ -0,0 +1,57 @@
package com.ruoyi.device.domain.impl.machine.instruction;
/**
* 抽象指令基类提供默认实现和下游节点管理
*/
public abstract class AbstractInstruction implements Instruction {
/**
* 成功后执行的下一个指令
*/
private Instruction onSuccessInstruction;
/**
* 失败后执行的下一个指令
*/
private Instruction onFailureInstruction;
@Override
public Instruction getOnSuccessInstruction() {
return onSuccessInstruction;
}
@Override
public Instruction getOnFailureInstruction() {
return onFailureInstruction;
}
/**
* 设置成功后执行的指令支持链式调用
*/
public <T extends AbstractInstruction> T onSuccess(Instruction instruction) {
this.onSuccessInstruction = instruction;
return (T) this;
}
/**
* 设置失败后执行的指令支持链式调用
*/
public <T extends AbstractInstruction> T onFailure(Instruction instruction) {
this.onFailureInstruction = instruction;
return (T) this;
}
/**
* 设置无论成功失败都执行的指令支持链式调用
*/
public <T extends AbstractInstruction> T then(Instruction instruction) {
this.onFailureInstruction = instruction;
this.onSuccessInstruction = instruction;
return (T) this;
}
}

View File

@ -0,0 +1,150 @@
package com.ruoyi.device.domain.impl.machine.instruction;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.function.Predicate;
/**
* 回调配置用于方法回调和状态回调
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class CallbackConfig {
/**
* 回调类型枚举
*/
public enum CallbackType {
/**
* 方法回调设备对指令的直接响应
* - 收到匹配的响应 成功
* - 收到不匹配的响应 失败立即
* - 超时 失败
*/
METHOD,
/**
* 状态回调等待设备状态变化
* - 收到匹配的状态 成功
* - 收到不匹配的状态 继续等待
* - 超时 失败
*/
STATE
}
/**
* 监听的MQTT主题
*/
private String topic;
/**
* 字段路径支持嵌套 "data.status"
*/
private String fieldPath;
/**
* 期望的字段值
*/
private Object expectedValue;
/**
* 自定义判断逻辑如果设置则优先使用此逻辑
*/
private Predicate<Object> customPredicate;
/**
* 超时时间毫秒
*/
@Builder.Default
private long timeoutMs = 10000;
/**
* 事务ID字段路径用于匹配回调消息 "tid"
*/
private String tidFieldPath;
/**
* 业务ID字段路径用于匹配回调消息 "bid"
*/
private String bidFieldPath;
/**
* 期望的事务ID值从InstructionContext中获取
*/
private String expectedTid;
/**
* 期望的业务ID值从InstructionContext中获取
*/
private String expectedBid;
/**
* 回调类型由框架自动设置不需要手动指定
* - getMethodCallbackConfig() 返回的配置会被设置为 METHOD
* - getStateCallbackConfig() 返回的配置会被设置为 STATE
*/
private CallbackType callbackType;
/**
* 判断消息是否匹配
* 注意tid/bid 的匹配已经在 MqttCallbackRegistry 注册层完成这里只检查业务字段
*/
public boolean matches(Object messageBody) {
if (customPredicate != null) {
return customPredicate.test(messageBody);
}
// 检查业务字段是否匹配
Object fieldValue = extractFieldValue(messageBody, fieldPath);
return expectedValue == null || expectedValue.equals(fieldValue);
}
/**
* 从消息体中提取字段值
*/
private Object extractFieldValue(Object messageBody, String path) {
if (messageBody == null || path == null) {
return null;
}
// 如果 messageBody 是字符串尝试解析为 JSON
Object current = messageBody;
if (messageBody instanceof String) {
try {
ObjectMapper objectMapper = new ObjectMapper();
current = objectMapper.readValue((String) messageBody, Object.class);
} catch (Exception e) {
// 解析失败返回 null
return null;
}
}
String[] parts = path.split("\\.");
for (String part : parts) {
if (current == null) {
return null;
}
if (current instanceof java.util.Map) {
current = ((java.util.Map<?, ?>) current).get(part);
} else {
try {
java.lang.reflect.Field field = current.getClass().getDeclaredField(part);
field.setAccessible(true);
current = field.get(current);
} catch (Exception e) {
return null;
}
}
}
return current;
}
}

View File

@ -0,0 +1,87 @@
package com.ruoyi.device.domain.impl.machine.instruction;
/**
* 指令接口
* 一个指令包含四个部分
* a. 判断是否可以执行该指令
* b. 执行远程调用如MQTT发送
* c. 等待方法回调并判断该方法是否成功执行
* d. 等待状态回调并判断结果是否OK
*/
public interface Instruction {
/**
* 获取指令名称
*/
String getName();
/**
* a. 判断是否可以执行该指令
*/
default boolean canExecute(InstructionContext context){
return true;
}
/**
* b. 执行远程调用如MQTT发送
*/
default void executeRemoteCall(InstructionContext context) throws Exception{
}
/**
* c. 获取方法回调配置可选
* 返回null表示不需要方法回调
*/
CallbackConfig getMethodCallbackConfig(InstructionContext context);
/**
* d. 获取状态回调配置可选
* 返回null表示不需要状态回调
*/
CallbackConfig getStateCallbackConfig(InstructionContext context);
/**
* 获取指令超时时间毫秒
*/
default long getTimeoutMs() {
return 1000; // 默认10秒
}
/**
* 指令执行完成回调无论成功失败都会调用
*/
default void onComplete(InstructionContext context, InstructionResult result) {
// 默认空实现
}
/**
* 获取成功后执行的下一个指令
*/
default Instruction getOnSuccessInstruction() {
return null;
}
/**
* 获取失败后执行的下一个指令
*/
default Instruction getOnFailureInstruction() {
return null;
}
/**
* 根据执行结果获取下一个指令
*
* @param success 是否成功
* @return 下一个指令如果没有则返回null
*/
default Instruction getNextInstruction(boolean success) {
// 根据成功失败返回对应的指令
if (success) {
return getOnSuccessInstruction();
} else {
return getOnFailureInstruction();
}
}
}

View File

@ -0,0 +1,85 @@
package com.ruoyi.device.domain.impl.machine.instruction;
import com.ruoyi.device.domain.impl.machine.mqtt.MqttClient;
import lombok.Data;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
/**
* 指令执行上下文
*/
@Data
public class InstructionContext {
/**
* 设备SN号
*/
private String sn;
/**
* 厂家类型
*/
private String vendorType;
/**
* MQTT客户端用于发送MQTT消息
*/
private MqttClient mqttClient;
/**
* 上下文数据用于在指令间传递数据
*/
private Map<String, Object> contextData = new HashMap<>();
/**
* 命令参数
*/
private Map<String, Object> commandParams = new HashMap<>();
/**
* 事务IDTransaction ID- 用于匹配回调消息
* 在命令执行阶段生成用于标识本次指令执行
*/
private String tid;
/**
* 业务IDBusiness ID- 用于匹配回调消息
* 在命令执行阶段生成用于标识本次业务操作
*/
private String bid;
public InstructionContext(String sn, String vendorType) {
this.sn = sn;
this.vendorType = vendorType;
// 自动生成 tid bid
this.tid = UUID.randomUUID().toString();
this.bid = UUID.randomUUID().toString();
}
public InstructionContext(String sn, String vendorType, MqttClient mqttClient) {
this.sn = sn;
this.vendorType = vendorType;
this.mqttClient = mqttClient;
// 自动生成 tid bid
this.tid = UUID.randomUUID().toString();
this.bid = UUID.randomUUID().toString();
}
public void putContextData(String key, Object value) {
contextData.put(key, value);
}
public Object getContextData(String key) {
return contextData.get(key);
}
public void putCommandParam(String key, Object value) {
commandParams.put(key, value);
}
public Object getCommandParam(String key) {
return commandParams.get(key);
}
}

View File

@ -0,0 +1,44 @@
package com.ruoyi.device.domain.impl.machine.instruction;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 指令执行结果
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class InstructionResult {
/**
* 是否成功
*/
private boolean success;
/**
* 错误信息
*/
private String errorMessage;
/**
* 结果数据
*/
private Object data;
public static InstructionResult success() {
return new InstructionResult(true, null, null);
}
public static InstructionResult success(Object data) {
return new InstructionResult(true, null, data);
}
public static InstructionResult failure(String errorMessage) {
return new InstructionResult(false, errorMessage, null);
}
public static InstructionResult timeout() {
return new InstructionResult(false, "指令执行超时", null);
}
}

View File

@ -0,0 +1,45 @@
package com.ruoyi.device.domain.impl.machine.mqtt;
import lombok.AllArgsConstructor;
import lombok.Data;
import java.util.function.Consumer;
/**
* MQTT回调处理器
*/
@Data
@AllArgsConstructor
public class MqttCallbackHandler {
/**
* 回调ID用于取消注册
*/
private String callbackId;
/**
* 监听的主题
*/
private String topic;
/**
* 消息处理器
*/
private Consumer<Object> messageHandler;
/**
* 超时时间毫秒
*/
private long timeoutMs;
/**
* 注册时间
*/
private long registerTime;
/**
* 是否已超时
*/
public boolean isTimeout() {
return System.currentTimeMillis() - registerTime > timeoutMs;
}
}

View File

@ -0,0 +1,361 @@
package com.ruoyi.device.domain.impl.machine.mqtt;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.device.domain.impl.machine.mqtt.store.MqttCallbackInfo;
import com.ruoyi.device.domain.impl.machine.mqtt.store.MqttCallbackStore;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.net.InetAddress;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
/**
* MQTT回调注册中心
* 用于注册和管理MQTT消息的回调处理器,他的 handleMessage 需要被真实的MQTT回调去调用
*
* 架构说明
* - 回调元数据存储在 MqttCallbackStore 支持内存/Redis
* - Consumer<Object> 回调函数存储在本地内存中无法序列化
* - 多节点部署时通过 Redis Pub/Sub 在节点间传递消息
*/
@Slf4j
@Component
public class MqttCallbackRegistry {
/**
* 回调存储层支持内存Redis等多种实现
*/
private final MqttCallbackStore callbackStore;
/**
* 回调ID -> 本地消息处理器Consumer 无法序列化只能存储在本地
*/
private final Map<String, Consumer<Object>> localHandlers = new ConcurrentHashMap<>();
/**
* 当前节点ID用于 Redis Pub/Sub 路由
*/
private String nodeId;
/**
* ObjectMapper 用于序列化消息
*/
private final ObjectMapper objectMapper = new ObjectMapper();
@Value("${machine.node.id:#{null}}")
private String configuredNodeId;
public MqttCallbackRegistry(MqttCallbackStore callbackStore) {
this.callbackStore = callbackStore;
}
@PostConstruct
public void init() {
// 初始化节点ID
if (configuredNodeId != null && !configuredNodeId.isEmpty()) {
nodeId = configuredNodeId;
} else {
// 自动生成节点ID主机名 + UUID
try {
String hostname = InetAddress.getLocalHost().getHostName();
nodeId = hostname + "-" + UUID.randomUUID().toString().substring(0, 8);
} catch (Exception e) {
nodeId = "node-" + UUID.randomUUID().toString().substring(0, 8);
}
}
// 订阅当前节点的消息用于 Redis Pub/Sub
callbackStore.subscribeNodeMessages(nodeId, this::handleNodeMessage);
log.info("MQTT回调注册中心初始化完成节点ID: {}, 存储实现: {}",
nodeId, callbackStore.getClass().getSimpleName());
}
/**
* 注册回调
*
* @param topic 监听的主题
* @param messageHandler 消息处理器
* @param timeoutMs 超时时间毫秒
* @return 回调ID用于取消注册
*/
public String registerCallback(String topic, Consumer<Object> messageHandler, long timeoutMs) {
return registerCallback(topic, messageHandler, timeoutMs, null, null, null, null);
}
/**
* 注册回调支持 tid/bid 过滤
*
* @param topic 监听的主题
* @param messageHandler 消息处理器
* @param timeoutMs 超时时间毫秒
* @param tidFieldPath tid 字段路径 "tid"
* @param expectedTid 期望的 tid
* @param bidFieldPath bid 字段路径 "bid"
* @param expectedBid 期望的 bid
* @return 回调ID用于取消注册
*/
public String registerCallback(String topic, Consumer<Object> messageHandler, long timeoutMs,
String tidFieldPath, String expectedTid,
String bidFieldPath, String expectedBid) {
String callbackId = UUID.randomUUID().toString();
// 1. 创建回调信息并存储到存储层
MqttCallbackInfo callbackInfo = MqttCallbackInfo.builder()
.callbackId(callbackId)
.topic(topic)
.timeoutMs(timeoutMs)
.registerTime(System.currentTimeMillis())
.nodeId(nodeId)
.tidFieldPath(tidFieldPath)
.expectedTid(expectedTid)
.bidFieldPath(bidFieldPath)
.expectedBid(expectedBid)
.build();
callbackStore.registerCallback(callbackInfo);
// 2. Consumer 存储到本地内存
localHandlers.put(callbackId, messageHandler);
log.debug("注册MQTT回调: callbackId={}, topic={}, timeoutMs={}, nodeId={}, tid={}, bid={}",
callbackId, topic, timeoutMs, nodeId, expectedTid, expectedBid);
return callbackId;
}
/**
* 取消注册回调
*
* @param callbackId 回调ID
*/
public void unregisterCallback(String callbackId) {
// 1. 从存储层删除回调信息
callbackStore.unregisterCallback(callbackId);
// 2. 从本地内存删除 Consumer
localHandlers.remove(callbackId);
log.debug("取消注册MQTT回调: callbackId={}", callbackId);
}
/**
* 处理接收到的MQTT消息由真实的 MQTT 客户端调用
*
* @param topic 主题
* @param messageBody 消息体
*/
public void handleMessage(String topic, Object messageBody) {
// 1. 从存储层获取所有等待该 topic 的回调信息
List<MqttCallbackInfo> callbacks = callbackStore.getCallbacksByTopic(topic);
if (callbacks.isEmpty()) {
return;
}
log.debug("处理MQTT消息: topic={}, callbackCount={}", topic, callbacks.size());
// 2. 序列化消息体用于跨节点传递
String messageBodyJson;
try {
messageBodyJson = objectMapper.writeValueAsString(messageBody);
} catch (Exception e) {
log.error("序列化消息体失败: topic={}", topic, e);
return;
}
// 3. 处理每个回调
for (MqttCallbackInfo callbackInfo : callbacks) {
try {
// 检查是否超时
if (callbackInfo.isTimeout()) {
log.warn("MQTT回调已超时: callbackId={}, topic={}",
callbackInfo.getCallbackId(), topic);
unregisterCallback(callbackInfo.getCallbackId());
continue;
}
// 检查 tid/bid 是否匹配如果配置了
if (!matchesTidBid(callbackInfo, messageBody)) {
log.debug("MQTT消息 tid/bid 不匹配,跳过回调: callbackId={}, topic={}",
callbackInfo.getCallbackId(), topic);
continue;
}
// 判断回调是在本节点还是其他节点
if (nodeId.equals(callbackInfo.getNodeId())) {
// 本节点的回调直接执行
executeLocalCallback(callbackInfo.getCallbackId(), messageBody);
} else {
// 其他节点的回调通过 Redis Pub/Sub 转发
callbackStore.publishMessageToNode(
callbackInfo.getNodeId(),
callbackInfo.getCallbackId(),
messageBodyJson
);
log.debug("转发消息到节点: nodeId={}, callbackId={}",
callbackInfo.getNodeId(), callbackInfo.getCallbackId());
}
} catch (Exception e) {
log.error("处理MQTT回调失败: callbackId={}, topic={}",
callbackInfo.getCallbackId(), topic, e);
}
}
}
/**
* 检查消息的 tid/bid 是否匹配
*
* @param callbackInfo 回调信息
* @param messageBody 消息体
* @return true 如果匹配或未配置 tid/bidfalse 如果不匹配
*/
private boolean matchesTidBid(MqttCallbackInfo callbackInfo, Object messageBody) {
// 1. 检查 tid 是否匹配如果配置了
if (callbackInfo.getTidFieldPath() != null && callbackInfo.getExpectedTid() != null) {
Object tidValue = extractFieldValue(messageBody, callbackInfo.getTidFieldPath());
if (!callbackInfo.getExpectedTid().equals(tidValue)) {
log.debug("tid 不匹配: expected={}, actual={}", callbackInfo.getExpectedTid(), tidValue);
return false; // tid 不匹配
}
}
// 2. 检查 bid 是否匹配如果配置了
if (callbackInfo.getBidFieldPath() != null && callbackInfo.getExpectedBid() != null) {
Object bidValue = extractFieldValue(messageBody, callbackInfo.getBidFieldPath());
if (!callbackInfo.getExpectedBid().equals(bidValue)) {
log.debug("bid 不匹配: expected={}, actual={}", callbackInfo.getExpectedBid(), bidValue);
return false; // bid 不匹配
}
}
// 3. tid/bid 都匹配或未配置
return true;
}
/**
* 从消息体中提取字段值
*
* @param messageBody 消息体
* @param fieldPath 字段路径支持嵌套 "data.status"
* @return 字段值
*/
private Object extractFieldValue(Object messageBody, String fieldPath) {
if (messageBody == null || fieldPath == null) {
return null;
}
// 如果 messageBody 是字符串尝试解析为 JSON
Object current = messageBody;
if (messageBody instanceof String) {
try {
current = objectMapper.readValue((String) messageBody, Object.class);
} catch (Exception e) {
log.warn("解析消息体失败: {}", messageBody);
return null;
}
}
String[] parts = fieldPath.split("\\.");
for (String part : parts) {
if (current == null) {
return null;
}
if (current instanceof Map) {
current = ((Map<?, ?>) current).get(part);
} else {
try {
java.lang.reflect.Field field = current.getClass().getDeclaredField(part);
field.setAccessible(true);
current = field.get(current);
} catch (Exception e) {
log.debug("提取字段失败: fieldPath={}, part={}", fieldPath, part);
return null;
}
}
}
return current;
}
/**
* 执行本地回调
*
* @param callbackId 回调ID
* @param messageBody 消息体
*/
private void executeLocalCallback(String callbackId, Object messageBody) {
Consumer<Object> handler = localHandlers.get(callbackId);
if (handler != null) {
try {
handler.accept(messageBody);
log.debug("执行本地回调成功: callbackId={}", callbackId);
} catch (Exception e) {
log.error("执行本地回调失败: callbackId={}", callbackId, e);
}
} else {
log.warn("本地回调处理器不存在: callbackId={}", callbackId);
}
}
/**
* 处理从 Redis Pub/Sub 接收到的节点消息
*
* @param callbackId 回调ID
* @param messageBodyJson 消息体JSON 字符串
*/
private void handleNodeMessage(String callbackId, String messageBodyJson) {
try {
// 反序列化消息体
Object messageBody = objectMapper.readValue(messageBodyJson, Object.class);
// 执行本地回调
executeLocalCallback(callbackId, messageBody);
} catch (Exception e) {
log.error("处理节点消息失败: callbackId={}", callbackId, e);
}
}
/**
* 清理超时的回调
*/
public void cleanupTimeoutCallbacks() {
List<MqttCallbackInfo> allCallbacks = callbackStore.getAllCallbacks();
for (MqttCallbackInfo callbackInfo : allCallbacks) {
if (callbackInfo.isTimeout()) {
log.warn("清理超时的MQTT回调: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic());
unregisterCallback(callbackInfo.getCallbackId());
}
}
}
/**
* 获取当前注册的回调数量
*/
public int getCallbackCount() {
return localHandlers.size();
}
/**
* 清理所有回调仅用于测试环境
* 警告此方法会清理所有回调包括未超时的仅应在测试环境中使用
*/
public void cleanupAllCallbacks() {
List<MqttCallbackInfo> allCallbacks = callbackStore.getAllCallbacks();
for (MqttCallbackInfo callbackInfo : allCallbacks) {
log.debug("清理MQTT回调: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic());
unregisterCallback(callbackInfo.getCallbackId());
}
log.info("已清理所有MQTT回调共{}个", allCallbacks.size());
}
}

View File

@ -0,0 +1,14 @@
package com.ruoyi.device.domain.impl.machine.mqtt;
import org.springframework.stereotype.Component;
/**
* MQTT客户端
*/
@Component
public class MqttClient {
public void sendMessage(String topic, String message) {
}
}

View File

@ -0,0 +1,85 @@
package com.ruoyi.device.domain.impl.machine.mqtt.store;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
/**
* 基于内存的 MQTT 回调存储实现
* 适用于单节点部署或开发测试环境
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "memory", matchIfMissing = true)
public class InMemoryMqttCallbackStore implements MqttCallbackStore {
/**
* 主题 -> 回调信息列表
*/
private final Map<String, CopyOnWriteArrayList<MqttCallbackInfo>> topicCallbacks = new ConcurrentHashMap<>();
/**
* 回调ID -> 回调信息
*/
private final Map<String, MqttCallbackInfo> callbackMap = new ConcurrentHashMap<>();
@Override
public void registerCallback(MqttCallbackInfo callbackInfo) {
topicCallbacks.computeIfAbsent(callbackInfo.getTopic(), k -> new CopyOnWriteArrayList<>())
.add(callbackInfo);
callbackMap.put(callbackInfo.getCallbackId(), callbackInfo);
log.debug("注册MQTT回调到内存: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic());
}
@Override
public void unregisterCallback(String callbackId) {
MqttCallbackInfo callbackInfo = callbackMap.remove(callbackId);
if (callbackInfo != null) {
CopyOnWriteArrayList<MqttCallbackInfo> callbacks = topicCallbacks.get(callbackInfo.getTopic());
if (callbacks != null) {
callbacks.remove(callbackInfo);
if (callbacks.isEmpty()) {
topicCallbacks.remove(callbackInfo.getTopic());
}
}
log.debug("从内存中取消注册MQTT回调: callbackId={}, topic={}",
callbackId, callbackInfo.getTopic());
}
}
@Override
public List<MqttCallbackInfo> getCallbacksByTopic(String topic) {
CopyOnWriteArrayList<MqttCallbackInfo> callbacks = topicCallbacks.get(topic);
return callbacks != null ? new ArrayList<>(callbacks) : new ArrayList<>();
}
@Override
public MqttCallbackInfo getCallbackById(String callbackId) {
return callbackMap.get(callbackId);
}
@Override
public List<MqttCallbackInfo> getAllCallbacks() {
return new ArrayList<>(callbackMap.values());
}
@Override
public void publishMessageToNode(String nodeId, String callbackId, String messageBody) {
// 内存实现中不需要跨节点通信此方法为空操作
log.trace("内存实现不需要发布消息到节点: nodeId={}, callbackId={}", nodeId, callbackId);
}
@Override
public void subscribeNodeMessages(String nodeId, NodeMessageListener messageListener) {
// 内存实现中不需要订阅节点消息此方法为空操作
log.trace("内存实现不需要订阅节点消息: nodeId={}", nodeId);
}
}

View File

@ -0,0 +1,72 @@
package com.ruoyi.device.domain.impl.machine.mqtt.store;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* MQTT 回调信息可序列化到 Redis
* 不包含 Consumer只包含回调的元数据
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MqttCallbackInfo implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 回调ID用于取消注册
*/
private String callbackId;
/**
* 监听的主题
*/
private String topic;
/**
* 超时时间毫秒
*/
private long timeoutMs;
/**
* 注册时间
*/
private long registerTime;
/**
* 注册该回调的节点ID用于 Redis Pub/Sub 路由
*/
private String nodeId;
/**
* 事务ID字段路径用于匹配回调消息 "tid"
*/
private String tidFieldPath;
/**
* 期望的事务ID值
*/
private String expectedTid;
/**
* 业务ID字段路径用于匹配回调消息 "bid"
*/
private String bidFieldPath;
/**
* 期望的业务ID值
*/
private String expectedBid;
/**
* 是否已超时
*/
public boolean isTimeout() {
return System.currentTimeMillis() - registerTime > timeoutMs;
}
}

View File

@ -0,0 +1,79 @@
package com.ruoyi.device.domain.impl.machine.mqtt.store;
import java.util.List;
/**
* MQTT 回调存储接口
* 提供回调信息的存储和获取抽象支持多种实现内存Redis等
*/
public interface MqttCallbackStore {
/**
* 注册回调信息
*
* @param callbackInfo 回调信息
*/
void registerCallback(MqttCallbackInfo callbackInfo);
/**
* 取消注册回调
*
* @param callbackId 回调ID
*/
void unregisterCallback(String callbackId);
/**
* 根据 topic 获取所有等待该 topic 的回调信息
*
* @param topic MQTT 主题
* @return 回调信息列表
*/
List<MqttCallbackInfo> getCallbacksByTopic(String topic);
/**
* 根据 callbackId 获取回调信息
*
* @param callbackId 回调ID
* @return 回调信息如果不存在返回 null
*/
MqttCallbackInfo getCallbackById(String callbackId);
/**
* 获取所有回调信息用于清理超时回调
*
* @return 所有回调信息列表
*/
List<MqttCallbackInfo> getAllCallbacks();
/**
* 发布消息到指定节点用于 Redis Pub/Sub
* 在内存实现中此方法为空操作
*
* @param nodeId 节点ID
* @param callbackId 回调ID
* @param messageBody 消息体JSON 字符串
*/
void publishMessageToNode(String nodeId, String callbackId, String messageBody);
/**
* 订阅当前节点的消息用于 Redis Pub/Sub
* 在内存实现中此方法为空操作
*
* @param nodeId 当前节点ID
* @param messageListener 消息监听器
*/
void subscribeNodeMessages(String nodeId, NodeMessageListener messageListener);
/**
* 节点消息监听器
*/
interface NodeMessageListener {
/**
* 处理接收到的消息
*
* @param callbackId 回调ID
* @param messageBody 消息体JSON 字符串
*/
void onMessage(String callbackId, String messageBody);
}
}

View File

@ -0,0 +1,267 @@
package com.ruoyi.device.domain.impl.machine.mqtt.store;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* 基于 Redis MQTT 回调存储实现
* 适用于多节点部署的生产环境
*
* 架构说明
* 1. 回调信息存储在 Redis Hash mqtt:callback:{callbackId} -> MqttCallbackInfo (JSON)
* 2. Topic 索引存储在 Redis Set mqtt:topic:{topic} -> Set<callbackId>
* 3. 使用 Redis Pub/Sub 在节点间传递 MQTT 消息mqtt:node:{nodeId} -> {callbackId, messageBody}
*
* 工作流程
* - 节点A 注册回调 -> 存储到 Redis
* - 节点B 收到 MQTT 消息 -> Redis 查询等待该 topic 的回调 -> 通过 Pub/Sub 发送到对应节点
* - 节点A 收到 Pub/Sub 消息 -> 执行本地的 Consumer 回调
*
* 使用方式
* 1. application.properties 中配置machine.state.store.type=redis
* 2. 配置 Redis 连接信息
* 3. 实现 Redis 相关的序列化和 Pub/Sub 逻辑
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "redis")
public class RedisMqttCallbackStore implements MqttCallbackStore {
private final StringRedisTemplate stringRedisTemplate;
private final RedisMessageListenerContainer redisMessageListenerContainer;
private final ObjectMapper objectMapper;
// Redis key 前缀
private static final String CALLBACK_KEY_PREFIX = "mqtt:callback:";
private static final String TOPIC_INDEX_PREFIX = "mqtt:topic:";
private static final String NODE_CHANNEL_PREFIX = "mqtt:node:";
// 配置回调信息的过期时间
private static final long EXPIRE_SECONDS = 3600; // 1小时
/**
* Lua 脚本注册 MQTT 回调
* 使用 Lua 脚本保证原子性避免竞态条件
*
* KEYS[1]: Topic 索引 key (mqtt:topic:{topic})
* KEYS[2]: 回调信息 key (mqtt:callback:{callbackId})
* ARGV[1]: callbackId
* ARGV[2]: 过期时间
* ARGV[3]: 回调信息 JSON
*
* 返回值: 1 表示成功
*/
private static final String REGISTER_CALLBACK_SCRIPT =
"redis.call('SADD', KEYS[1], ARGV[1]) " +
"redis.call('EXPIRE', KEYS[1], ARGV[2]) " +
"redis.call('SETEX', KEYS[2], ARGV[2], ARGV[3]) " +
"return 1";
/**
* Lua 脚本取消注册 MQTT 回调
* 使用 Lua 脚本保证原子性
*
* KEYS[1]: Topic 索引 key (mqtt:topic:{topic})
* KEYS[2]: 回调信息 key (mqtt:callback:{callbackId})
* ARGV[1]: callbackId
*
* 返回值: 1 表示成功
*/
private static final String UNREGISTER_CALLBACK_SCRIPT =
"redis.call('SREM', KEYS[1], ARGV[1]) " +
"redis.call('DEL', KEYS[2]) " +
"return 1";
public RedisMqttCallbackStore(
StringRedisTemplate stringRedisTemplate,
@Qualifier("machineFrameworkRedisMessageListenerContainer") RedisMessageListenerContainer redisMessageListenerContainer,
ObjectMapper objectMapper) {
this.stringRedisTemplate = stringRedisTemplate;
this.redisMessageListenerContainer = redisMessageListenerContainer;
this.objectMapper = objectMapper;
log.info("使用 Redis MQTT 回调存储实现");
}
@Override
public void registerCallback(MqttCallbackInfo callbackInfo) {
try {
// 1. 序列化回调信息为 JSON
String json = objectMapper.writeValueAsString(callbackInfo);
// 2. 准备 Redis key
String topicKey = TOPIC_INDEX_PREFIX + callbackInfo.getTopic();
String callbackKey = CALLBACK_KEY_PREFIX + callbackInfo.getCallbackId();
// 3. 使用 Lua 脚本原子性地注册回调
// 先添加到 Topic 索引再存储回调信息避免竞态条件
stringRedisTemplate.execute(
new DefaultRedisScript<>(REGISTER_CALLBACK_SCRIPT, Long.class),
Arrays.asList(topicKey, callbackKey),
callbackInfo.getCallbackId(),
String.valueOf(EXPIRE_SECONDS),
json
);
log.debug("注册MQTT回调到Redis: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic());
} catch (JsonProcessingException e) {
log.error("序列化回调信息失败: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic(), e);
throw new RuntimeException("注册MQTT回调失败: 序列化错误", e);
} catch (Exception e) {
log.error("注册MQTT回调到Redis失败: callbackId={}, topic={}",
callbackInfo.getCallbackId(), callbackInfo.getTopic(), e);
// 不抛出异常让上层通过超时机制处理
// 这样可以避免因为 Redis 临时故障导致整个命令执行失败
}
}
@Override
public void unregisterCallback(String callbackId) {
try {
// 1. 获取回调信息需要知道 topic 才能删除索引
MqttCallbackInfo callbackInfo = getCallbackById(callbackId);
if (callbackInfo == null) {
log.debug("回调信息不存在,无需取消注册: callbackId={}", callbackId);
return;
}
// 2. 准备 Redis key
String topicKey = TOPIC_INDEX_PREFIX + callbackInfo.getTopic();
String callbackKey = CALLBACK_KEY_PREFIX + callbackId;
// 3. 使用 Lua 脚本原子性地取消注册回调
stringRedisTemplate.execute(
new DefaultRedisScript<>(UNREGISTER_CALLBACK_SCRIPT, Long.class),
Arrays.asList(topicKey, callbackKey),
callbackId
);
log.debug("从Redis中取消注册MQTT回调: callbackId={}, topic={}",
callbackId, callbackInfo.getTopic());
} catch (Exception e) {
log.error("从Redis中取消注册MQTT回调失败: callbackId={}", callbackId, e);
// 不抛出异常取消注册失败不影响主流程
// 回调会因为 TTL 自动过期
}
}
@Override
public List<MqttCallbackInfo> getCallbacksByTopic(String topic) {
// 1. Topic 索引获取所有 callbackId
String topicKey = TOPIC_INDEX_PREFIX + topic;
Set<String> callbackIds = stringRedisTemplate.opsForSet().members(topicKey);
if (callbackIds == null || callbackIds.isEmpty()) {
return new ArrayList<>();
}
// 2. 批量获取回调信息
List<MqttCallbackInfo> callbacks = new ArrayList<>();
for (String callbackId : callbackIds) {
MqttCallbackInfo callbackInfo = getCallbackById(callbackId);
if (callbackInfo != null) {
callbacks.add(callbackInfo);
}
}
return callbacks;
}
@Override
public MqttCallbackInfo getCallbackById(String callbackId) {
String callbackKey = CALLBACK_KEY_PREFIX + callbackId;
String json = stringRedisTemplate.opsForValue().get(callbackKey);
if (json == null) {
return null;
}
try {
return objectMapper.readValue(json, MqttCallbackInfo.class);
} catch (JsonProcessingException e) {
log.error("反序列化回调信息失败: callbackId={}", callbackId, e);
return null;
}
}
@Override
public List<MqttCallbackInfo> getAllCallbacks() {
// 1. 扫描所有 mqtt:callback:* key
Set<String> keys = stringRedisTemplate.keys(CALLBACK_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) {
return new ArrayList<>();
}
// 2. 批量获取回调信息
List<MqttCallbackInfo> callbacks = new ArrayList<>();
for (String key : keys) {
String callbackId = key.substring(CALLBACK_KEY_PREFIX.length());
MqttCallbackInfo callbackInfo = getCallbackById(callbackId);
if (callbackInfo != null) {
callbacks.add(callbackInfo);
}
}
return callbacks;
}
@Override
public void publishMessageToNode(String nodeId, String callbackId, String messageBody) {
try {
// 1. 构造消息体
Map<String, String> message = new HashMap<>();
message.put("callbackId", callbackId);
message.put("messageBody", messageBody);
// 2. 序列化消息
String json = objectMapper.writeValueAsString(message);
// 3. 发布到节点频道
String channel = NODE_CHANNEL_PREFIX + nodeId;
stringRedisTemplate.convertAndSend(channel, json);
log.debug("发布消息到节点: nodeId={}, callbackId={}, channel={}",
nodeId, callbackId, channel);
} catch (JsonProcessingException e) {
log.error("序列化节点消息失败: nodeId={}, callbackId={}", nodeId, callbackId, e);
}
}
@Override
public void subscribeNodeMessages(String nodeId, NodeMessageListener messageListener) {
// 1. 创建消息监听器
MessageListener redisMessageListener = (message, pattern) -> {
try {
String json = new String(message.getBody());
Map<String, String> data = objectMapper.readValue(json,
new TypeReference<Map<String, String>>() {});
String callbackId = data.get("callbackId");
String messageBody = data.get("messageBody");
messageListener.onMessage(callbackId, messageBody);
} catch (Exception e) {
log.error("处理Redis Pub/Sub消息失败", e);
}
};
// 2. 订阅节点频道
String channel = NODE_CHANNEL_PREFIX + nodeId;
redisMessageListenerContainer.addMessageListener(redisMessageListener, new ChannelTopic(channel));
log.info("订阅节点消息: nodeId={}, channel={}", nodeId, channel);
}
}

View File

@ -0,0 +1,218 @@
生产需要实现
MysqlSnVendorMappingRepository 这个类
MQTT回调的地方需要转发到MqttCallbackRegistry 这个类
需要实现 MqttClient 这边消息发送的逻辑
需要配置 sn.repository.type=mysql
单节点部署(默认)
# 使用内存存储(默认配置)
machine.state.store.type=memory
# 以下配置生产需要修改为mysql同时实现 MysqlSnVendorMappingRepository 这个类
sn.repository.type=memory
多节点部署
# 切换到 Redis 存储
machine.state.store.type=redis
# 配置节点ID可选不配置会自动生成
machine.node.id=node-1
#本地启动redis
#docker run --name some-redis -d -p 6379:6379 redi
# Redis 配置
spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=your-password
┌─────────────────────────────────────────────────────────────────┐
│ 步骤1: 节点A 执行命令并注册回调 │
└─────────────────────────────────────────────────────────────────┘
节点A: executeCommand(TAKE_OFF)
节点A: registerCallback(topic="dji/SN9527/response", ...)
节点A: MqttCallbackStore.registerCallback()
Redis: 存储回调信息(使用两个 Key 的原因:性能优化)
【Key 1】回调详细信息Hash 结构)
- Key: mqtt:callback:{callbackId}
- Value: {callbackId, topic, nodeId="nodeA", timeoutMs, registerTime, ...}
- 作用: 存储单个回调的完整信息
- 查询: O(1) 时间复杂度,通过 callbackId 直接获取
【Key 2】Topic 索引Set 结构)
- Key: mqtt:topic:dji/SN9527/response
- Value: Set<callbackId> // 例如: ["abc-123", "def-456", "ghi-789"]
- 作用: 快速查询等待某个 topic 的所有回调
- 查询: O(1) 时间复杂度,直接获取 callbackId 列表
【为什么需要两个 Key
如果只用一个 Key 存储所有回调,查询时需要遍历所有回调并过滤 topic
时间复杂度为 O(n)。使用 Topic 索引后,可以直接获取目标回调列表,
时间复杂度降为 O(1),大幅提升性能。
【示例】
假设有 3 个回调:
- callbackId="abc-123", topic="dji/SN9527/response", nodeId="nodeA"
- callbackId="def-456", topic="dji/SN9527/state", nodeId="nodeB"
- callbackId="ghi-789", topic="dji/SN9527/response", nodeId="nodeA"
Redis 存储结构:
mqtt:callback:abc-123 → {callbackId:"abc-123", topic:"dji/SN9527/response", nodeId:"nodeA"}
mqtt:callback:def-456 → {callbackId:"def-456", topic:"dji/SN9527/state", nodeId:"nodeB"}
mqtt:callback:ghi-789 → {callbackId:"ghi-789", topic:"dji/SN9527/response", nodeId:"nodeA"}
mqtt:topic:dji/SN9527/response → ["abc-123", "ghi-789"]
mqtt:topic:dji/SN9527/state → ["def-456"]
查询 topic="dji/SN9527/response" 的回调:
1. 从索引获取: SMEMBERS mqtt:topic:dji/SN9527/response → ["abc-123", "ghi-789"]
2. 批量获取详情: MGET mqtt:callback:abc-123 mqtt:callback:ghi-789
3. 总耗时: O(1) + O(k)k 是该 topic 的回调数量(通常很小)
【Redis 数据清理时机】
Redis 中的回调数据有两种清理机制:
┌─────────────────────────────────────────────────────────────┐
│ 1⃣ 主动清理(业务逻辑触发) │
└─────────────────────────────────────────────────────────────┘
触发时机:
✅ 回调成功执行后TransactionExecutor 的 finally 块)
✅ 回调超时后TransactionExecutor 的 finally 块)
✅ handleMessage 检测到超时(转发前检查)
清理操作:
unregisterCallback(callbackId)
1. 获取回调信息: GET mqtt:callback:{callbackId}
2. 删除回调信息: DEL mqtt:callback:{callbackId}
3. 从索引中移除: SREM mqtt:topic:{topic} {callbackId}
示例:
T0: 注册回调,超时时间 10 秒
T5: 收到 MQTT 响应,回调执行成功
T5: 立即清理 Redis 数据 ✅
- DEL mqtt:callback:abc-123
- SREM mqtt:topic:dji/SN9527/response abc-123
┌─────────────────────────────────────────────────────────────┐
│ 2⃣ 被动清理Redis TTL 自动过期) │
└─────────────────────────────────────────────────────────────┘
作用:兜底机制,防止异常情况下的数据残留
设置方式:
// 注册回调时设置 TTL
SET mqtt:callback:{callbackId} {json} EX 3600 // 1小时后自动过期
EXPIRE mqtt:topic:{topic} 3600 // 1小时后自动过期
触发时机:
⚠️ 应用异常崩溃,主动清理未执行
⚠️ 网络分区,无法删除 Redis 数据
⚠️ 代码 Bug主动清理失败
示例:
T0: 注册回调TTL=3600秒1小时
T5: 应用崩溃,主动清理未执行 ❌
T3600: Redis 自动删除过期数据 ✅
- mqtt:callback:abc-123 自动过期删除
- mqtt:topic:dji/SN9527/response 自动过期删除
【推荐配置】
TTL 应该设置为回调超时时间的 2-3 倍,例如:
- 回调超时: 10 秒
- Redis TTL: 30 秒10秒 × 3
这样可以确保:
✅ 正常情况下,主动清理会在 10 秒内完成
✅ 异常情况下Redis 会在 30 秒后自动清理
✅ 避免设置过长的 TTL 导致内存浪费
【注意事项】
⚠️ Topic 索引的 TTL 问题:
如果同一个 topic 有多个回调,每次添加新回调时都会刷新 TTL。
这可能导致索引的 TTL 比单个回调的 TTL 更长。
解决方案:
方案1: 不为 Topic 索引设置 TTL只在删除最后一个 callbackId 时删除索引
方案2: 每次查询时过滤掉已过期的 callbackId推荐
节点A: 本地内存存储 Consumer<Object>
- localHandlers.put(callbackId, consumer)
节点A: 订阅 Redis Pub/Sub 频道
- Channel: mqtt:node:nodeA
┌─────────────────────────────────────────────────────────────────┐
│ 步骤2: MQTT Broker 将响应路由到节点B不是节点A
└─────────────────────────────────────────────────────────────────┘
MQTT Broker: 收到设备响应
MQTT Broker: 将消息路由到节点B随机/轮询)
节点B: MqttCallbackRegistry.handleMessage(topic, messageBody)
┌─────────────────────────────────────────────────────────────────┐
│ 步骤3: 节点B 从 Redis 查询等待该 topic 的回调 │
└─────────────────────────────────────────────────────────────────┘
节点B: callbackStore.getCallbacksByTopic("dji/SN9527/response")
Redis: 查询 mqtt:topic:dji/SN9527/response
Redis: 返回 Set<callbackId>
Redis: 批量获取回调信息
- mqtt:callback:{callbackId1} → {nodeId="nodeA", ...}
- mqtt:callback:{callbackId2} → {nodeId="nodeA", ...}
节点B: 获得回调列表 List<MqttCallbackInfo>
┌─────────────────────────────────────────────────────────────────┐
│ 步骤4: 节点B 判断回调属于哪个节点 │
└─────────────────────────────────────────────────────────────────┘
节点B: for (MqttCallbackInfo callback : callbacks) {
if (nodeId.equals(callback.getNodeId())) {
// 本节点的回调,直接执行
executeLocalCallback(...)
} else {
// 其他节点的回调,转发到目标节点
callbackStore.publishMessageToNode(...)
}
}
┌─────────────────────────────────────────────────────────────────┐
│ 步骤5: 节点B 通过 Redis Pub/Sub 转发消息到节点A │
└─────────────────────────────────┘
节点B: callbackStore.publishMessageToNode(
nodeId="nodeA",
callbackId="xxx",
messageBody="{...}" // JSON 字符串
)
Redis Pub/Sub: PUBLISH mqtt:node:nodeA
{
"callbackId": "xxx",
"messageBody": "{...}"
}
┌─────────────────────────────────────────────────────────────────┐
│ 步骤6: 节点A 收到 Redis Pub/Sub 消息 │
└─────────────────────────────────────────────────────────────────┘
节点A: Redis Pub/Sub Listener 收到消息
节点A: handleNodeMessage(callbackId, messageBodyJson)
节点A: 反序列化消息体
- Object messageBody = objectMapper.readValue(messageBodyJson)
节点A: executeLocalCallback(callbackId, messageBody)
节点A: 从本地内存获取 Consumer
- Consumer<Object> handler = localHandlers.get(callbackId)
节点A: 执行回调
- handler.accept(messageBody)
✅ 命令执行成功!

View File

@ -0,0 +1,15 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 机巢状态枚举
*/
public enum AirportState {
/**
* 未知状态服务器重启后的初始状态等待第一次心跳同步,同时也是离线状态
*/
UNKNOWN,
/**
* 在线
*/
ONLINE
}

View File

@ -0,0 +1,22 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 舱门状态枚举
*/
public enum CoverState {
/**
* 未知状态服务器重启后的初始状态等待第一次心跳同步
*/
UNKNOWN,
/**
* 舱门已关闭
*/
CLOSED,
/**
* 舱门已打开
*/
OPENED,
}

View File

@ -0,0 +1,21 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 调试模式状态枚举
*/
public enum DebugModeState {
/**
* 未知状态
*/
UNKNOWN,
/**
* 调试模式
*/
ENTERED,
/**
* 退出调试模式
*/
EXITED
}

View File

@ -0,0 +1,23 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 飞行控制模式DRC状态枚举
*/
public enum DrcState {
/**
* 未知状态服务器重启后的初始状态等待第一次心跳同步
*/
UNKNOWN,
/**
* 退出状态DRC模式已退出
*/
EXITED,
/**
* 进入状态已进入DRC模式
*/
ENTERED,
}

View File

@ -0,0 +1,33 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 无人机状态枚举
* 分为准备中 -> 飞行中 -> 返航 三个大状态
*/
public enum DroneState {
/**
* 未知状态服务器重启后的初始状态等待第一次心跳同步,同时也是离线状态
*/
UNKNOWN,
/**
* 在线
*/
ONLINE,
/**
* 飞行中
*/
FLYING,
/**
* 到达目的地
*/
ARRIVED,
/**
* 返航中
*/
RETURNING,
}

View File

@ -0,0 +1,50 @@
package com.ruoyi.device.domain.impl.machine.state;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 设备的六套大状态
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MachineStates {
/**
* 无人机状态
*/
private DroneState droneState = DroneState.UNKNOWN;
/**
* 机巢状态
*/
private AirportState airportState = AirportState.UNKNOWN;
/**
* 舱门状态
*/
private CoverState coverState = CoverState.UNKNOWN;
/**
* DRC状态
*/
private DrcState drcState = DrcState.UNKNOWN;
/**
* 调试模式状态
*/
private DebugModeState debugModeState = DebugModeState.UNKNOWN;
/**
* 急停状态
*/
private StopState stopState = StopState.UNKNOWN;
/**
* 复制当前状态
*/
public MachineStates copy() {
return new MachineStates(droneState, airportState, coverState, drcState, debugModeState, stopState);
}
}

View File

@ -0,0 +1,22 @@
package com.ruoyi.device.domain.impl.machine.state;
/**
* 急停状态
*/
public enum StopState {
/**
* 未知状态服务器重启后的初始状态等待第一次心跳同步
*/
UNKNOWN,
/**
* 退出状态
*/
EXITED,
/**
* 进入状态
*/
ENTERED,
}

View File

@ -0,0 +1,195 @@
package com.ruoyi.device.domain.impl.machine.statemachine;
import com.ruoyi.device.domain.impl.machine.state.*;
import com.ruoyi.device.domain.impl.machine.statemachine.store.MachineStateStore;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 设备状态管理器
* 负责状态的业务逻辑处理和变化通知底层存储由 MachineStateStore 实现
*/
@Slf4j
@Component
public class MachineStateManager {
/**
* 状态存储层支持内存Redis等多种实现
*/
private final MachineStateStore stateStore;
/**
* 状态变化监听器
*/
private final Map<String, StateChangeListener> stateChangeListeners = new ConcurrentHashMap<>();
public MachineStateManager(MachineStateStore stateStore) {
this.stateStore = stateStore;
log.info("设备状态管理器初始化完成,使用存储实现: {}", stateStore.getClass().getSimpleName());
}
/**
* 获取设备状态
*/
public MachineStates getStates(String sn) {
return stateStore.getStates(sn);
}
/**
* 设置无人机状态
*/
public void setDroneState(String sn, DroneState newState) {
MachineStates states = getStates(sn);
DroneState oldState = states.getDroneState();
if (oldState != newState) {
states.setDroneState(newState);
stateStore.saveStates(sn, states); // 保存到存储层
log.info("无人机状态变化: sn={}, {} -> {}", sn, oldState, newState);
notifyStateChange(sn, states.copy());
}
}
/**
* 设置机巢状态
*/
public void setAirportState(String sn, AirportState newState) {
MachineStates states = getStates(sn);
AirportState oldState = states.getAirportState();
if (oldState != newState) {
states.setAirportState(newState);
stateStore.saveStates(sn, states); // 保存到存储层
log.info("机巢状态变化: sn={}, {} -> {}", sn, oldState, newState);
notifyStateChange(sn, states.copy());
}
}
/**
* 设置舱门状态
*/
public void setCoverState(String sn, CoverState newState) {
MachineStates states = getStates(sn);
CoverState oldState = states.getCoverState();
if (oldState != newState) {
states.setCoverState(newState);
stateStore.saveStates(sn, states); // 保存到存储层
log.info("舱门状态变化: sn={}, {} -> {}", sn, oldState, newState);
notifyStateChange(sn, states.copy());
}
}
/**
* 设置DRC状态
*/
public void setDrcState(String sn, DrcState newState) {
MachineStates states = getStates(sn);
DrcState oldState = states.getDrcState();
if (oldState != newState) {
states.setDrcState(newState);
stateStore.saveStates(sn, states); // 保存到存储层
log.info("DRC状态变化: sn={}, {} -> {}", sn, oldState, newState);
notifyStateChange(sn, states.copy());
}
}
/**
* 批量更新状态用于心跳同步
* 只更新非UNKNOWN的状态避免覆盖已有状态
*/
public void updateStates(String sn, MachineStates newStates) {
updateStates(sn, newStates, false);
}
/**
* 批量更新状态用于心跳同步
*
* @param sn 设备SN号
* @param newStates 新状态
* @param forceUpdate 是否强制更新所有状态包括UNKNOWN
*/
public void updateStates(String sn, MachineStates newStates, boolean forceUpdate) {
MachineStates currentStates = getStates(sn);
boolean changed = false;
// 更新无人机状态如果不是UNKNOWN或强制更新
if (forceUpdate || newStates.getDroneState() != DroneState.UNKNOWN) {
if (currentStates.getDroneState() != newStates.getDroneState()) {
currentStates.setDroneState(newStates.getDroneState());
changed = true;
}
}
// 更新机巢状态如果不是UNKNOWN或强制更新
if (forceUpdate || newStates.getAirportState() != AirportState.UNKNOWN) {
if (currentStates.getAirportState() != newStates.getAirportState()) {
currentStates.setAirportState(newStates.getAirportState());
changed = true;
}
}
// 更新舱门状态如果不是UNKNOWN或强制更新
if (forceUpdate || newStates.getCoverState() != CoverState.UNKNOWN) {
if (currentStates.getCoverState() != newStates.getCoverState()) {
currentStates.setCoverState(newStates.getCoverState());
changed = true;
}
}
// 更新DRC状态如果不是UNKNOWN或强制更新
if (forceUpdate || newStates.getDrcState() != DrcState.UNKNOWN) {
if (currentStates.getDrcState() != newStates.getDrcState()) {
currentStates.setDrcState(newStates.getDrcState());
changed = true;
}
}
if (changed) {
stateStore.saveStates(sn, currentStates); // 保存到存储层
log.info("设备状态批量更新: sn={}, states={}", sn, currentStates);
notifyStateChange(sn, currentStates.copy());
}
}
/**
* 注册状态变化监听器
*/
public void registerStateChangeListener(String listenerId, StateChangeListener listener) {
stateChangeListeners.put(listenerId, listener);
log.debug("注册状态变化监听器: listenerId={}", listenerId);
}
/**
* 取消注册状态变化监听器
*/
public void unregisterStateChangeListener(String listenerId) {
stateChangeListeners.remove(listenerId);
log.debug("取消注册状态变化监听器: listenerId={}", listenerId);
}
/**
* 通知状态变化
*/
private void notifyStateChange(String sn, MachineStates newStates) {
for (StateChangeListener listener : stateChangeListeners.values()) {
try {
listener.onStateChange(sn, newStates);
} catch (Exception e) {
log.error("状态变化监听器执行失败: sn={}", sn, e);
}
}
}
/**
* 移除设备状态设备下线时调用
*/
public void removeStates(String sn) {
stateStore.removeStates(sn);
log.info("移除设备状态: sn={}", sn);
}
}

View File

@ -0,0 +1,18 @@
package com.ruoyi.device.domain.impl.machine.statemachine;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
/**
* 状态变化监听器
*/
@FunctionalInterface
public interface StateChangeListener {
/**
* 状态变化回调
*
* @param sn 设备SN号
* @param newStates 新状态
*/
void onStateChange(String sn, MachineStates newStates);
}

View File

@ -0,0 +1,50 @@
package com.ruoyi.device.domain.impl.machine.statemachine.store;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 基于内存的设备状态存储实现
* 适用于单节点部署或开发测试环境
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "memory", matchIfMissing = true)
public class InMemoryMachineStateStore implements MachineStateStore {
/**
* SN -> 设备状态
*/
private final Map<String, MachineStates> stateMap = new ConcurrentHashMap<>();
@Override
public MachineStates getStates(String sn) {
return stateMap.computeIfAbsent(sn, k -> {
log.debug("创建新的设备状态: sn={}", sn);
return new MachineStates();
});
}
@Override
public void saveStates(String sn, MachineStates states) {
stateMap.put(sn, states);
log.debug("保存设备状态到内存: sn={}", sn);
}
@Override
public void removeStates(String sn) {
stateMap.remove(sn);
log.debug("从内存中移除设备状态: sn={}", sn);
}
@Override
public boolean exists(String sn) {
return stateMap.containsKey(sn);
}
}

View File

@ -0,0 +1,43 @@
package com.ruoyi.device.domain.impl.machine.statemachine.store;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
/**
* 设备状态存储接口
* 提供状态的存储和获取抽象支持多种实现内存Redis等
*/
public interface MachineStateStore {
/**
* 获取设备状态
* 如果设备状态不存在返回一个新的默认状态对象
*
* @param sn 设备SN号
* @return 设备状态
*/
MachineStates getStates(String sn);
/**
* 保存设备状态
*
* @param sn 设备SN号
* @param states 设备状态
*/
void saveStates(String sn, MachineStates states);
/**
* 移除设备状态
*
* @param sn 设备SN号
*/
void removeStates(String sn);
/**
* 检查设备状态是否存在
*
* @param sn 设备SN号
* @return 是否存在
*/
boolean exists(String sn);
}

View File

@ -0,0 +1,93 @@
package com.ruoyi.device.domain.impl.machine.statemachine.store;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* 基于 Redis 的设备状态存储实现
* 适用于多节点部署的生产环境
*
* Redis 数据结构
* - Key: machine:state:{sn}
* - Value: MachineStates (JSON)
* - TTL: 86400 24小时
*
* 使用方式
* 1. application.properties 中配置machine.state.store.type=redis
* 2. 配置 Redis 连接信息
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "redis")
public class RedisMachineStateStore implements MachineStateStore {
private final StringRedisTemplate redisTemplate;
private final ObjectMapper objectMapper;
// Redis key 前缀
private static final String KEY_PREFIX = "machine:state:";
// 配置状态的过期时间
private static final long EXPIRE_SECONDS = 86400; // 24小时
public RedisMachineStateStore(StringRedisTemplate redisTemplate, ObjectMapper objectMapper) {
this.redisTemplate = redisTemplate;
this.objectMapper = objectMapper;
log.info("使用 Redis 设备状态存储实现");
}
@Override
public MachineStates getStates(String sn) {
String key = KEY_PREFIX + sn;
String json = redisTemplate.opsForValue().get(key);
if (json == null) {
log.debug("Redis 中不存在设备状态,返回默认状态: sn={}", sn);
MachineStates states = new MachineStates();
saveStates(sn, states);
return states;
}
try {
MachineStates states = objectMapper.readValue(json, MachineStates.class);
log.debug("从 Redis 获取设备状态: sn={}", sn);
return states;
} catch (JsonProcessingException e) {
log.error("反序列化设备状态失败: sn={}", sn, e);
return new MachineStates();
}
}
@Override
public void saveStates(String sn, MachineStates states) {
try {
String key = KEY_PREFIX + sn;
String json = objectMapper.writeValueAsString(states);
redisTemplate.opsForValue().set(key, json, EXPIRE_SECONDS, TimeUnit.SECONDS);
log.debug("保存设备状态到 Redis: sn={}", sn);
} catch (JsonProcessingException e) {
log.error("序列化设备状态失败: sn={}", sn, e);
}
}
@Override
public void removeStates(String sn) {
String key = KEY_PREFIX + sn;
redisTemplate.delete(key);
log.debug("从 Redis 中移除设备状态: sn={}", sn);
}
@Override
public boolean exists(String sn) {
String key = KEY_PREFIX + sn;
return Boolean.TRUE.equals(redisTemplate.hasKey(key));
}
}

View File

@ -0,0 +1,49 @@
package com.ruoyi.device.domain.impl.machine.vendor;
import com.ruoyi.device.domain.impl.machine.command.CommandType;
import com.ruoyi.device.domain.impl.machine.command.Transaction;
import com.ruoyi.device.domain.impl.machine.state.MachineStates;
import java.util.List;
import java.util.Map;
/**
* 厂家配置接口
* 每个厂家需要实现此接口定义其支持的命令和状态转换规则
*/
public interface VendorConfig {
/**
* 获取厂家类型
*/
String getVendorType();
/**
* 获取厂家名称
*/
String getVendorName();
/**
* 获取指定命令的事务定义
*
* @param commandType 命令类型
* @return 事务定义如果不支持该命令则返回null
*/
Transaction getTransaction(CommandType commandType);
/**
* 判断在当前状态下是否可以执行指定命令
*
* @param currentStates 当前状态
* @param commandType 命令类型
* @return 是否可以执行
*/
boolean canExecuteCommand(MachineStates currentStates, CommandType commandType);
/**
* 获取在当前状态下可以执行的命令列表
*
* @param currentStates 当前状态
* @return 可执行的命令列表
*/
List<CommandType> getAvailableCommands(MachineStates currentStates);
}

View File

@ -0,0 +1,119 @@
package com.ruoyi.device.domain.impl.machine.vendor;
import com.ruoyi.device.domain.impl.machine.vendor.store.SnVendorMappingStore;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 厂家注册中心
* 管理厂家配置和 SN 到厂家的映射关系
*
* 架构说明
* - 厂家配置存储在本地内存vendorConfigs
* - SN 到厂家类型的映射通过 SnVendorMappingStore 存储支持内存/Redis+MySQL
*/
@Slf4j
@Component
public class VendorRegistry {
/**
* 厂家类型 -> 厂家配置存储在本地内存
*/
private final Map<String, VendorConfig> vendorConfigs = new ConcurrentHashMap<>();
/**
* SN 到厂家类型的映射存储层支持内存Redis+MySQL等多种实现
*/
private final SnVendorMappingStore mappingStore;
public VendorRegistry(SnVendorMappingStore mappingStore) {
this.mappingStore = mappingStore;
log.info("厂家注册中心初始化完成,使用映射存储实现: {}", mappingStore.getClass().getSimpleName());
}
/**
* 注册厂家配置
*/
public void registerVendor(VendorConfig vendorConfig) {
vendorConfigs.put(vendorConfig.getVendorType(), vendorConfig);
log.info("注册厂家配置: vendorType={}, vendorName={}",
vendorConfig.getVendorType(), vendorConfig.getVendorName());
}
/**
* 绑定SN到厂家
*
* @deprecated 不建议直接调用此方法应该通过数据库或配置文件管理 SN 映射关系
*/
@Deprecated
public void bindSnToVendor(String sn, String vendorType) {
if (!vendorConfigs.containsKey(vendorType)) {
throw new IllegalArgumentException("未注册的厂家类型: " + vendorType);
}
mappingStore.saveMapping(sn, vendorType);
log.debug("绑定SN到厂家: sn={}, vendorType={}", sn, vendorType);
}
/**
* 解绑SN
*
* @deprecated 不建议直接调用此方法应该通过数据库或配置文件管理 SN 映射关系
*/
@Deprecated
public void unbindSn(String sn) {
mappingStore.removeMapping(sn);
log.debug("解绑SN: sn={}", sn);
}
/**
* 获取SN对应的厂家类型
*
* 查询顺序Redis+MySQL 模式
* 1. 先从 Redis 缓存获取
* 2. Redis 没有则从 MySQL 数据库获取
* 3. 获取到后存入 Redis 缓存
*/
public String getVendorType(String sn) {
return mappingStore.getVendorType(sn);
}
/**
* 获取SN对应的厂家配置
*
* 查询顺序Redis+MySQL 模式
* 1. 通过 mappingStore 获取厂家类型Redis MySQL 缓存
* 2. 根据厂家类型获取厂家配置
*/
public VendorConfig getVendorConfig(String sn) {
String vendorType = getVendorType(sn);
if (vendorType == null) {
return null;
}
return vendorConfigs.get(vendorType);
}
/**
* 根据厂家类型获取厂家配置
*/
public VendorConfig getVendorConfigByType(String vendorType) {
return vendorConfigs.get(vendorType);
}
/**
* 判断SN是否已绑定厂家
*/
public boolean isSnBound(String sn) {
return mappingStore.exists(sn);
}
/**
* 获取所有已注册的厂家类型
*/
public java.util.Set<String> getAllVendorTypes() {
return vendorConfigs.keySet();
}
}

View File

@ -0,0 +1,121 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji;
import com.ruoyi.device.domain.impl.machine.command.CommandType;
import com.ruoyi.device.domain.impl.machine.command.Transaction;
import com.ruoyi.device.domain.impl.machine.state.*;
import com.ruoyi.device.domain.impl.machine.vendor.VendorConfig;
import com.ruoyi.device.domain.impl.machine.vendor.dji.instruction.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* 大疆无人机厂家配置
*/
@Slf4j
@Component
public class DjiVendorConfig implements VendorConfig {
private final Map<CommandType, Transaction> transactionMap = new HashMap<>();
public DjiVendorConfig() {
initTransactions();
}
@Override
public String getVendorType() {
return "DJI";
}
@Override
public String getVendorName() {
return "大疆";
}
@Override
public Transaction getTransaction(CommandType commandType) {
return transactionMap.get(commandType);
}
@Override
public boolean canExecuteCommand(MachineStates currentStates, CommandType commandType) {
DroneState droneState = currentStates.getDroneState();
AirportState airportState = currentStates.getAirportState();
CoverState coverState = currentStates.getCoverState();
DebugModeState debugModeState = currentStates.getDebugModeState();
StopState stopState = currentStates.getStopState();
switch (commandType) {
case TAKE_OFF:
return droneState == DroneState.ONLINE && airportState == AirportState.ONLINE;
case RETURN_HOME:
return droneState == DroneState.FLYING || droneState == DroneState.ARRIVED;
default:
return true;
}
}
@Override
public List<CommandType> getAvailableCommands(MachineStates currentStates) {
List<CommandType> availableCommands = new ArrayList<>();
for (CommandType commandType : CommandType.values()) {
if (canExecuteCommand(currentStates, commandType)) {
availableCommands.add(commandType);
}
}
return availableCommands;
}
/**
* 初始化事务定义
*/
private void initTransactions() {
// 起飞命令
Transaction takeOffTransaction = new Transaction("起飞", CommandType.TAKE_OFF)
.root(new DjiTakeOffInstruction())
.setTimeout(10000);
transactionMap.put(takeOffTransaction.getCommandType(), takeOffTransaction);
/**
* 开仓命令 Transaction
* 流程说明
* 1. root: 检查是否已在调试模式
* - 成功直接执行开仓命令
* - 失败先开启调试模式再执行开仓命令
*/
// 创建检查调试模式的指令root
DjiCheckDebugModeInstruction checkDebugMode = new DjiCheckDebugModeInstruction();
// 创建开仓指令成功分支
DjiOpenCoverInstruction openCoverAfterCheck = new DjiOpenCoverInstruction();
// 创建开启调试模式的指令失败分支
DjiEnableDebugModeInstruction enableDebugMode = new DjiEnableDebugModeInstruction();
// 创建开仓指令失败分支的子指令
DjiOpenCoverInstruction openCoverAfterEnable = new DjiOpenCoverInstruction();
// 构建指令树
checkDebugMode
.onSuccess(openCoverAfterCheck) // 如果已在调试模式直接开仓
.onFailure(enableDebugMode // 如果不在调试模式先开启调试模式
.onSuccess(openCoverAfterEnable)); // 开启调试模式成功后再开仓
Transaction openCoverTransaction = new Transaction("开仓", CommandType.OPEN_COVER)
.root(checkDebugMode)
.setTimeout(80000); // 总超时时间80秒
transactionMap.put(openCoverTransaction.getCommandType(), openCoverTransaction);
log.info("大疆厂家配置初始化完成,共配置{}个命令", transactionMap.size());
}
}

View File

@ -0,0 +1,57 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆取消指点指令
*/
@Slf4j
public class DjiCancelPointInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_CANCEL_POINT";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆取消指点指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"cancelPoint\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("cancelPoint")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("POINT_CANCELLED")
.timeoutMs(30000)
.build();
}
@Override
public long getTimeoutMs() {
return 30000;
}
}

View File

@ -0,0 +1,51 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆检查调试模式指令
* 用于判断设备是否已经进入调试模式
*/
@Slf4j
public class DjiCheckDebugModeInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_CHECK_DEBUG_MODE";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("检查大疆设备调试模式状态: sn={}", sn);
// 不需要发送命令只需要等待状态回调
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
// 不需要方法回调
return null;
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
// 等待设备状态回调判断是否在调试模式
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("debugMode")
.expectedValue("ENABLED")
.timeoutMs(3000) // 3秒超时如果没有收到说明不在调试模式
.build();
}
@Override
public long getTimeoutMs() {
return 3000;
}
}

View File

@ -0,0 +1,58 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆关闭舱门指令
*/
@Slf4j
public class DjiCloseCoverInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_CLOSE_COVER";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆关闭舱门指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"closeCover\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("closeCover")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("coverState")
.expectedValue("CLOSED")
.timeoutMs(60000)
.build();
}
@Override
public long getTimeoutMs() {
return 60000;
}
}

View File

@ -0,0 +1,61 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆急停指令
*/
@Slf4j
public class DjiEmergencyStopInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_EMERGENCY_STOP";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆急停指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"emergencyStop\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("emergencyStop")
.timeoutMs(5000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.customPredicate(state -> {
// 急停状态可能是 EMERGENCY_STOP RETURN_EMERGENCY_STOP
return "EMERGENCY_STOP".equals(state) || "RETURN_EMERGENCY_STOP".equals(state);
})
.timeoutMs(30000)
.build();
}
@Override
public long getTimeoutMs() {
return 30000;
}
}

View File

@ -0,0 +1,55 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆开启调试模式指令
*/
@Slf4j
public class DjiEnableDebugModeInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_ENABLE_DEBUG_MODE";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆开启调试模式指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"enableDebugMode\"}";
context.getMqttClient().sendMessage(topic, payload);
log.debug("MQTT发送成功: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
// 等待开启调试模式命令的ACK响应
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("enableDebugMode")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
// 不需要状态回调只要收到ACK就认为命令发送成功
return null;
}
@Override
public long getTimeoutMs() {
return 10000;
}
}

View File

@ -0,0 +1,59 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆降落指令
*/
@Slf4j
public class DjiLandInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_LAND";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆降落指令: sn={}", sn);
// TODO: 实际的MQTT发送逻辑
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"land\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("land")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("PREPARING")
.timeoutMs(60000)
.build();
}
@Override
public long getTimeoutMs() {
return 90000;
}
}

View File

@ -0,0 +1,60 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆打开舱门指令
*/
@Slf4j
public class DjiOpenCoverInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_OPEN_COVER";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆打开舱门指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"openCover\"}";
context.getMqttClient().sendMessage(topic, payload);
log.debug("MQTT发送成功: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("openCover")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("coverState")
.expectedValue("OPENED")
.timeoutMs(60000)
.build();
}
@Override
public long getTimeoutMs() {
return 60000;
}
}

View File

@ -0,0 +1,63 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆指点飞行指令
*/
@Slf4j
public class DjiPointFlyInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_POINT_FLY";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
Object latitude = context.getCommandParam("latitude");
Object longitude = context.getCommandParam("longitude");
Object altitude = context.getCommandParam("altitude");
log.info("发送大疆指点飞行指令: sn={}, lat={}, lon={}, alt={}", sn, latitude, longitude, altitude);
String topic = "dji/" + sn + "/command";
String payload = String.format("{\"cmd\":\"pointFly\",\"latitude\":%s,\"longitude\":%s,\"altitude\":%s}",
latitude, longitude, altitude);
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("pointFly")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("POINT_FLYING")
.timeoutMs(30000)
.build();
}
@Override
public long getTimeoutMs() {
return 90000;
}
}

View File

@ -0,0 +1,61 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆继续飞行指令
*/
@Slf4j
public class DjiResumeFlightInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_RESUME_FLIGHT";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆继续飞行指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"resumeFlight\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("resumeFlight")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.customPredicate(state -> {
// 继续飞行后可能变为 FLYING RETURNING
return "FLYING".equals(state) || "RETURNING".equals(state);
})
.timeoutMs(30000)
.build();
}
@Override
public long getTimeoutMs() {
return 60000;
}
}

View File

@ -0,0 +1,58 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆返航指令
*/
@Slf4j
public class DjiReturnHomeInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_RETURN_HOME";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆返航指令: sn={}", sn);
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"returnHome\"}";
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("returnHome")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("RETURNING")
.timeoutMs(60000)
.build();
}
@Override
public long getTimeoutMs() {
return 120000;
}
}

View File

@ -0,0 +1,60 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆开始航线任务指令
*/
@Slf4j
public class DjiStartMissionInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_START_MISSION";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
Object missionId = context.getCommandParam("missionId");
log.info("发送大疆开始航线任务指令: sn={}, missionId={}", sn, missionId);
String topic = "dji/" + sn + "/command";
String payload = String.format("{\"cmd\":\"startMission\",\"missionId\":\"%s\"}", missionId);
log.debug("MQTT发送: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("cmd")
.expectedValue("startMission")
.timeoutMs(10000)
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
String sn = context.getSn();
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("FLYING")
.timeoutMs(60000)
.build();
}
@Override
public long getTimeoutMs() {
return 120000;
}
}

View File

@ -0,0 +1,65 @@
package com.ruoyi.device.domain.impl.machine.vendor.dji.instruction;
import com.ruoyi.device.domain.impl.machine.instruction.AbstractInstruction;
import com.ruoyi.device.domain.impl.machine.instruction.CallbackConfig;
import com.ruoyi.device.domain.impl.machine.instruction.InstructionContext;
import lombok.extern.slf4j.Slf4j;
/**
* 大疆起飞指令
*/
@Slf4j
public class DjiTakeOffInstruction extends AbstractInstruction {
@Override
public String getName() {
return "DJI_TAKE_OFF";
}
@Override
public void executeRemoteCall(InstructionContext context) throws Exception {
String sn = context.getSn();
log.info("发送大疆起飞指令: sn={}", sn);
// 通过 context 获取 MqttClient 并发送消息
String topic = "dji/" + sn + "/command";
String payload = "{\"cmd\":\"takeoff\"}";
context.getMqttClient().sendMessage(topic, payload);
log.debug("MQTT发送成功: topic={}, payload={}", topic, payload);
}
@Override
public CallbackConfig getMethodCallbackConfig(InstructionContext context) {
// return null;
String sn = context.getSn();
// 方法回调等待起飞指令的ACK响应
return CallbackConfig.builder()
.topic("dji/" + sn + "/response")
.fieldPath("data.result")
.expectedValue("takeoff")
.timeoutMs(10000) // 10秒超时
.build();
}
@Override
public CallbackConfig getStateCallbackConfig(InstructionContext context) {
// return null;
String sn = context.getSn();
// 状态回调等待无人机状态变为飞行中
return CallbackConfig.builder()
.topic("dji/" + sn + "/state")
.fieldPath("droneState")
.expectedValue("FLYING")
.timeoutMs(10000) // 10秒超时
.build();
}
@Override
public long getTimeoutMs() {
return 10000; // 10秒总超时
}
}

View File

@ -0,0 +1,53 @@
package com.ruoyi.device.domain.impl.machine.vendor.repository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 基于内存的 SN 到厂家类型映射持久化存储实现
* 适用于单节点部署或开发测试环境
*
* 注意这是默认实现当没有配置数据库时使用
*
* 使用方式
* 1. 不配置 machine.sn.repository.type默认使用内存实现
* 2. 或在 application.properties 中配置machine.sn.repository.type=memory
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.sn.repository.type", havingValue = "memory", matchIfMissing = true)
public class InMemorySnVendorMappingRepository implements SnVendorMappingRepository {
/**
* SN -> 厂家类型模拟数据库存储
*/
private final Map<String, String> dataStore = new ConcurrentHashMap<>();
@Override
public String findVendorTypeBySn(String sn) {
String vendorType = dataStore.get(sn);
log.debug("从内存数据库查询 SN 映射: sn={}, vendorType={}", sn, vendorType);
return vendorType;
}
@Override
public void save(String sn, String vendorType) {
dataStore.put(sn, vendorType);
log.debug("保存 SN 映射到内存数据库: sn={}, vendorType={}", sn, vendorType);
}
@Override
public void delete(String sn) {
dataStore.remove(sn);
log.debug("从内存数据库删除 SN 映射: sn={}", sn);
}
@Override
public boolean exists(String sn) {
return dataStore.containsKey(sn);
}
}

View File

@ -0,0 +1,97 @@
package com.ruoyi.device.domain.impl.machine.vendor.repository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 基于 MySQL SN 到厂家类型映射持久化存储实现
* 适用于生产环境
*
* MySQL 表结构示例
* CREATE TABLE sn_vendor_mapping (
* sn VARCHAR(64) PRIMARY KEY COMMENT '设备SN号',
* vendor_type VARCHAR(32) NOT NULL COMMENT '厂家类型',
* created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
* updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
* INDEX idx_vendor_type (vendor_type)
* ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='SN到厂家类型映射表';
*
* 使用方式
* 1. application.properties 中配置machine.sn.repository.type=mysql
* 2. 配置 MySQL 数据源
* 3. 创建上述表结构
* 4. 实现下面的 CRUD 方法
*
* 注意当前为空实现需要在连接数据库后完善
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.sn.repository.type", havingValue = "mysql")
public class MysqlSnVendorMappingRepository implements SnVendorMappingRepository {
// TODO: 注入 JdbcTemplate MyBatis Mapper
// private final JdbcTemplate jdbcTemplate;
//
// public MysqlSnVendorMappingRepository(JdbcTemplate jdbcTemplate) {
// this.jdbcTemplate = jdbcTemplate;
// }
public MysqlSnVendorMappingRepository() {
log.warn("使用 MySQL SN 映射持久化存储实现,但当前为空实现,请在连接数据库后完善");
}
@Override
public String findVendorTypeBySn(String sn) {
// TODO: 实现从 MySQL 查询
// try {
// return jdbcTemplate.queryForObject(
// "SELECT vendor_type FROM sn_vendor_mapping WHERE sn = ?",
// String.class,
// sn
// );
// } catch (EmptyResultDataAccessException e) {
// log.debug("MySQL 中不存在 SN 映射: sn={}", sn);
// return null;
// }
log.warn("MySQL SN 映射持久化存储未实现,返回 null: sn={}", sn);
return null;
}
@Override
public void save(String sn, String vendorType) {
// TODO: 实现保存到 MySQL
// jdbcTemplate.update(
// "INSERT INTO sn_vendor_mapping (sn, vendor_type) VALUES (?, ?) " +
// "ON DUPLICATE KEY UPDATE vendor_type = ?, updated_at = CURRENT_TIMESTAMP",
// sn, vendorType, vendorType
// );
// log.debug("保存 SN 映射到 MySQL: sn={}, vendorType={}", sn, vendorType);
log.warn("MySQL SN 映射持久化存储未实现,跳过保存: sn={}, vendorType={}", sn, vendorType);
}
@Override
public void delete(String sn) {
// TODO: 实现从 MySQL 删除
// jdbcTemplate.update("DELETE FROM sn_vendor_mapping WHERE sn = ?", sn);
// log.debug("从 MySQL 删除 SN 映射: sn={}", sn);
log.warn("MySQL SN 映射持久化存储未实现,跳过删除: sn={}", sn);
}
@Override
public boolean exists(String sn) {
// TODO: 实现检查 MySQL 中是否存在
// Integer count = jdbcTemplate.queryForObject(
// "SELECT COUNT(*) FROM sn_vendor_mapping WHERE sn = ?",
// Integer.class,
// sn
// );
// return count != null && count > 0;
log.warn("MySQL SN 映射持久化存储未实现,返回 false: sn={}", sn);
return false;
}
}

View File

@ -0,0 +1,46 @@
package com.ruoyi.device.domain.impl.machine.vendor.repository;
/**
* SN 到厂家类型映射的持久化存储接口
* 提供数据库层的 CRUD 操作抽象支持多种实现内存MySQL等
*
* 职责说明
* - 这是持久化层数据库层的抽象
* - SnVendorMappingStore 的区别
* - SnVendorMappingStore包含缓存逻辑Redis + Repository
* - SnVendorMappingRepository纯粹的持久化存储数据库
*/
public interface SnVendorMappingRepository {
/**
* 从持久化存储中查询 SN 对应的厂家类型
*
* @param sn 设备SN号
* @return 厂家类型如果不存在返回 null
*/
String findVendorTypeBySn(String sn);
/**
* 保存 SN 到厂家类型的映射到持久化存储
* 如果已存在则更新
*
* @param sn 设备SN号
* @param vendorType 厂家类型
*/
void save(String sn, String vendorType);
/**
* 从持久化存储中删除 SN 的映射关系
*
* @param sn 设备SN号
*/
void delete(String sn);
/**
* 检查持久化存储中是否存在 SN 的映射关系
*
* @param sn 设备SN号
* @return 是否存在
*/
boolean exists(String sn);
}

View File

@ -0,0 +1,47 @@
package com.ruoyi.device.domain.impl.machine.vendor.store;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 基于内存的 SN 到厂家类型映射存储实现
* 适用于单节点部署或开发测试环境
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "memory", matchIfMissing = true)
public class InMemorySnVendorMappingStore implements SnVendorMappingStore {
/**
* SN -> 厂家类型
*/
private final Map<String, String> snToVendorMap = new ConcurrentHashMap<>();
@Override
public String getVendorType(String sn) {
String vendorType = snToVendorMap.get(sn);
log.debug("从内存获取 SN 映射: sn={}, vendorType={}", sn, vendorType);
return vendorType;
}
@Override
public void saveMapping(String sn, String vendorType) {
snToVendorMap.put(sn, vendorType);
log.debug("保存 SN 映射到内存: sn={}, vendorType={}", sn, vendorType);
}
@Override
public void removeMapping(String sn) {
snToVendorMap.remove(sn);
log.debug("从内存中移除 SN 映射: sn={}", sn);
}
@Override
public boolean exists(String sn) {
return snToVendorMap.containsKey(sn);
}
}

View File

@ -0,0 +1,117 @@
package com.ruoyi.device.domain.impl.machine.vendor.store;
import com.ruoyi.device.domain.impl.machine.vendor.repository.SnVendorMappingRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* 基于 Redis + Repository SN 到厂家类型映射存储实现
* 适用于多节点部署的生产环境
*
* 架构说明
* 1. Redis 作为缓存层提供快速查询
* 2. Repository 作为持久化层支持内存/MySQL等实现
* 3. 查询流程Redis Repository 回写 Redis
*
* Redis 数据结构
* - Key: machine:sn:vendor:{sn}
* - Value: {vendorType}
* - TTL: 86400 24小时
*
* 使用方式
* 1. application.properties 中配置machine.state.store.type=redis
* 2. 配置 Redis 连接信息
* 3. Repository 会根据配置自动选择实现内存/MySQL
* 4. 实现 Redis 的缓存逻辑
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "machine.state.store.type", havingValue = "redis")
public class RedisSnVendorMappingStore implements SnVendorMappingStore {
private final StringRedisTemplate redisTemplate;
/**
* 持久化存储层支持内存/MySQL等实现
*/
private final SnVendorMappingRepository repository;
// Redis key 前缀
private static final String KEY_PREFIX = "machine:sn:vendor:";
// 配置缓存过期时间
private static final long CACHE_EXPIRE_SECONDS = 86400; // 24小时
public RedisSnVendorMappingStore(StringRedisTemplate redisTemplate,
SnVendorMappingRepository repository) {
this.redisTemplate = redisTemplate;
this.repository = repository;
log.info("使用 Redis+Repository SN 映射存储实现");
log.info("持久化层实现: {}", repository.getClass().getSimpleName());
}
@Override
public String getVendorType(String sn) {
// 1. 先从 Redis 缓存获取
String key = KEY_PREFIX + sn;
String vendorType = redisTemplate.opsForValue().get(key);
if (vendorType != null) {
log.debug("从 Redis 缓存获取 SN 映射: sn={}, vendorType={}", sn, vendorType);
return vendorType;
}
// 2. Redis 没有 Repository 持久化层获取
vendorType = repository.findVendorTypeBySn(sn);
if (vendorType != null) {
// 3. 获取到后存入 Redis 缓存
redisTemplate.opsForValue().set(key, vendorType, CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS);
log.debug("从 Repository 获取 SN 映射并缓存到 Redis: sn={}, vendorType={}", sn, vendorType);
return vendorType;
}
log.debug("Repository 中不存在 SN 映射: sn={}", sn);
return null;
}
@Override
public void saveMapping(String sn, String vendorType) {
// 1. 保存到 Repository 持久化层
repository.save(sn, vendorType);
// 2. 保存到 Redis 缓存
String key = KEY_PREFIX + sn;
redisTemplate.opsForValue().set(key, vendorType, CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS);
log.debug("保存 SN 映射到 Redis+Repository: sn={}, vendorType={}", sn, vendorType);
}
@Override
public void removeMapping(String sn) {
// 1. Repository 删除
repository.delete(sn);
// 2. Redis 删除
String key = KEY_PREFIX + sn;
redisTemplate.delete(key);
log.debug("从 Redis+Repository 中移除 SN 映射: sn={}", sn);
}
@Override
public boolean exists(String sn) {
// 1. 先检查 Redis
String key = KEY_PREFIX + sn;
if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
return true;
}
// 2. 再检查 Repository
return repository.exists(sn);
}
}

View File

@ -0,0 +1,49 @@
package com.ruoyi.device.domain.impl.machine.vendor.store;
/**
* SN 到厂家类型映射的存储接口
* 提供 SN 与厂家类型映射关系的存储和获取抽象支持多种实现内存Redis+MySQL等
*
* 数据流
* 1. 先从 Redis 缓存获取
* 2. Redis 没有则从 MySQL 数据库获取
* 3. 获取到后存入 Redis 缓存
* 4. 都没有则返回 null
*/
public interface SnVendorMappingStore {
/**
* 获取 SN 对应的厂家类型
*
* 查询顺序
* 1. 先从 Redis 缓存获取
* 2. Redis 没有则从 MySQL 数据库获取
* 3. 获取到后存入 Redis 缓存
*
* @param sn 设备SN号
* @return 厂家类型如果不存在返回 null
*/
String getVendorType(String sn);
/**
* 保存 SN 到厂家类型的映射
*
* @param sn 设备SN号
* @param vendorType 厂家类型
*/
void saveMapping(String sn, String vendorType);
/**
* 删除 SN 的映射关系
*
* @param sn 设备SN号
*/
void removeMapping(String sn);
/**
* 检查 SN 是否已有映射关系
*
* @param sn 设备SN号
* @return 是否存在映射
*/
boolean exists(String sn);
}