yuhaiming/water
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: 设备删除增加用户校验、ACK 在线状态刷新改非阻塞、停止跟踪日志

Browse Source 
- AppDeviceServiceImpl#deleteWithValidByIds 校验设备归属当前用户，
  避免任意已登录用户越权删除其他用户的设备及关联数据
- MqttCommandAckService#refreshDeviceOnline 由 lock.lock 改为 tryLock，
  防止 MQTT 处理线程在状态刷新锁上排队堆积；状态写入幂等，跳过本次刷新
  由后续 ACK 兜底
- 从版本控制中移除 logs/ 历史日志，gitignore 已忽略该目录

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
yuhaiming
2026-06-23 15:36:38 +08:00
parent 5548512633
commit 694e32abbf
17 changed files with 73 additions and 665 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java
View File

@@ -75,10 +75,14 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
return;
}
ack.setDeviceNo(deviceNo);
withCommandLock(ack.getCommandId(), () -> {
boolean ackSaved = withCommandLock(ack.getCommandId(), () -> {
deletePending(ack.getCommandId());
RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
});
if (!ackSaved) {
log.warn("[MQTT] ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, ack.getCommandId());
return;
}
refreshDeviceOnline(deviceNo);
log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} 状态={}", deviceNo, ack.getCommandId(), ack.getStatus());
}
@@ -99,10 +103,14 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
ack.setStatus("1");
ack.setMessage(payload);
withCommandLock(commandId, () -> {
boolean ackSaved = withCommandLock(commandId, () -> {
deletePending(commandId);
RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
});
if (!ackSaved) {
log.warn("[MQTT] 非 JSON ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, commandId);
return;
}
log.info("[MQTT] 收到非 JSON 命令确认 设备编号={} 命令编号={} 消息体={}", deviceNo, commandId, payload);
}
@@ -177,8 +185,11 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
RLock lock = RedisUtils.getClient().getLock(DEVICE_STATUS_LOCK_PREFIX + deviceNo);
boolean locked = false;
try {
lock.lock(10, TimeUnit.SECONDS);
locked = true;
locked = lock.tryLock(0, 10, TimeUnit.SECONDS);
if (!locked) {
// 其他线程正在刷新，本次跳过；状态写入幂等，下条 ACK 仍会兜底刷新
return;
}
Map<String, Object> statusCache = new HashMap<>();
statusCache.put("deviceNo", deviceNo);
statusCache.put("status", "1");
@@ -193,6 +204,9 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
.set(AppDevice::getStatus, "1")
.eq(AppDevice::getDeviceNo, deviceNo)
);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo);
} finally {
if (locked && lock.isHeldByCurrentThread()) {
lock.unlock();
@@ -209,20 +223,23 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
pendingIds().remove(commandId);
}
private void withCommandLock(String commandId, Runnable action) {
private boolean withCommandLock(String commandId, Runnable action) {
RLock lock = RedisUtils.getClient().getLock(mqttProperties.getCommandAck().getRetryLockKeyPrefix() + commandId);
boolean locked = false;
try {
locked = lock.tryLock(0, mqttProperties.getCommandAck().getRetryLockTtlMs(), TimeUnit.MILLISECONDS);
if (!locked) {
return;
return false;
}
action.run();
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("[MQTT] 命令锁等待被中断 命令编号={}", commandId);
return false;
} catch (RuntimeException e) {
log.error("[MQTT] 命令锁内处理失败 命令编号={}", commandId, e);
return false;
} finally {
if (locked && lock.isHeldByCurrentThread()) {
lock.unlock();