{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"我会快速查看当前 git diff 的文件清单和统计，确认这次提交的主线，避免只根据截断 diff 猜测。"}]}}{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"item_1","name":"bash","input":{"command":"\"C:\\\\Users\\\\admin\\\\AppData\\\\Local\\\\Microsoft\\\\WindowsApps\\\\pwsh.exe\" -Command 'git diff --stat'","description":"Run \"C:\\\\Users\\\\admin\\\\AppData\\\\Local\\\\Microsoft\\\\WindowsApps\\\\pwsh.exe\""}}]}}

2026-06-17 10:07:56 +08:00
parent c74363afa4
commit 5548512633
37 changed files with 1301 additions and 177 deletions
--- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java
+++ b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java
@@ -36,6 +36,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
    private MqttClientManager mqttClientManager;
    private final MqttProperties mqttProperties;
    private final AppDeviceMapper appDeviceMapper;
+    private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:";

    public void savePending(DeviceCommand command) {
        RedisUtils.setCacheObject(pendingKey(command.getCommandId()), command, Duration.ofSeconds(mqttProperties.getCommandAck().getPendingTtlSeconds()));
@@ -46,7 +47,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
        if (StringUtils.isBlank(commandId)) {
            return;
        }
-        deletePending(commandId);
+        withCommandLock(commandId, () -> deletePending(commandId));
    }

    @Override
@@ -74,8 +75,10 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
            return;
        }
        ack.setDeviceNo(deviceNo);
-        deletePending(ack.getCommandId());
-        RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+        withCommandLock(ack.getCommandId(), () -> {
+            deletePending(ack.getCommandId());
+            RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+        });
        refreshDeviceOnline(deviceNo);
        log.info("[MQTT] 收到命令确认 设备编号={} 命令编号={} 状态={}", deviceNo, ack.getCommandId(), ack.getStatus());
    }
@@ -96,8 +99,10 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
        ack.setStatus("1");
        ack.setMessage(payload);

-        deletePending(commandId);
-        RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+        withCommandLock(commandId, () -> {
+            deletePending(commandId);
+            RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+        });
        log.info("[MQTT] 收到非 JSON 命令确认 设备编号={} 命令编号={} 消息体={}", deviceNo, commandId, payload);
    }

@@ -121,24 +126,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
    }

    private void retryCommandIfLocked(String commandId, long now) {
-        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;
-            }
-            retryCommand(commandId, now);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.warn("[MQTT] 命令重试锁等待被中断 命令编号={}", commandId);
-        } catch (RuntimeException e) {
-            log.error("[MQTT] 命令重试失败 命令编号={}", commandId, e);
-        } finally {
-            if (locked && lock.isHeldByCurrentThread()) {
-                lock.unlock();
-            }
-        }
+        withCommandLock(commandId, () -> retryCommand(commandId, now));
    }

    private void retryCommand(String commandId, long now) {
@@ -186,20 +174,30 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
    }

    private void refreshDeviceOnline(String deviceNo) {
-        Map<String, Object> statusCache = new HashMap<>();
-        statusCache.put("deviceNo", deviceNo);
-        statusCache.put("status", "1");
-        statusCache.put("lastReportTime", Instant.now().toString());
-        RedisUtils.setCacheObject(
-            mqttProperties.getCommandAck().getDeviceStatusCachePrefix() + deviceNo,
-            statusCache,
-            Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds())
-        );
-        appDeviceMapper.update(null,
-            new LambdaUpdateWrapper<AppDevice>()
-                .set(AppDevice::getStatus, "1")
-                .eq(AppDevice::getDeviceNo, deviceNo)
-        );
+        RLock lock = RedisUtils.getClient().getLock(DEVICE_STATUS_LOCK_PREFIX + deviceNo);
+        boolean locked = false;
+        try {
+            lock.lock(10, TimeUnit.SECONDS);
+            locked = true;
+            Map<String, Object> statusCache = new HashMap<>();
+            statusCache.put("deviceNo", deviceNo);
+            statusCache.put("status", "1");
+            statusCache.put("lastReportTime", Instant.now().toString());
+            RedisUtils.setCacheObject(
+                mqttProperties.getCommandAck().getDeviceStatusCachePrefix() + deviceNo,
+                statusCache,
+                Duration.ofSeconds(mqttProperties.getCommandAck().getDeviceStatusCacheTtlSeconds())
+            );
+            appDeviceMapper.update(null,
+                new LambdaUpdateWrapper<AppDevice>()
+                    .set(AppDevice::getStatus, "1")
+                    .eq(AppDevice::getDeviceNo, deviceNo)
+            );
+        } finally {
+            if (locked && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+        }
    }

    private RSet<String> pendingIds() {
@@ -211,6 +209,27 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
        pendingIds().remove(commandId);
    }

+    private void 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;
+            }
+            action.run();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("[MQTT] 命令锁等待被中断 命令编号={}", commandId);
+        } catch (RuntimeException e) {
+            log.error("[MQTT] 命令锁内处理失败 命令编号={}", commandId, e);
+        } finally {
+            if (locked && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+        }
+    }
+
    private String pendingKey(String commandId) {
        return mqttProperties.getCommandAck().getPendingKeyPrefix() + commandId;
    }
--- a/water-common/water-common-oss/src/main/java/org/dromara/common/oss/core/OssClient.java
+++ b/water-common/water-common-oss/src/main/java/org/dromara/common/oss/core/OssClient.java
@@ -149,7 +149,7 @@ public class OssClient {
                            .contentType(contentType)
                            // 用于设置对象的访问控制列表（ACL）。不同云厂商对ACL的支持和实现方式有所不同，
                            // 因此根据具体的云服务提供商，你可能需要进行不同的配置（自行开启，阿里云有acl权限配置，腾讯云没有acl权限配置）
-                            //.acl(getAccessPolicy().getObjectCannedACL())
+                            .acl(getAccessPolicy().getObjectCannedACL())
                            .build()
                    );
                    if (log.isDebugEnabled()) {
@@ -203,7 +203,7 @@ public class OssClient {
                            .contentType(contentType)
                            // 用于设置对象的访问控制列表（ACL）。不同云厂商对ACL的支持和实现方式有所不同，
                            // 因此根据具体的云服务提供商，你可能需要进行不同的配置（自行开启，阿里云有acl权限配置，腾讯云没有acl权限配置）
-                            //.acl(getAccessPolicy().getObjectCannedACL())
+                            .acl(getAccessPolicy().getObjectCannedACL())
                            .build()
                    );
                    if (log.isDebugEnabled()) {
--- a/water-common/water-common-oss/src/main/java/org/dromara/common/oss/enums/AccessPolicyType.java
+++ b/water-common/water-common-oss/src/main/java/org/dromara/common/oss/enums/AccessPolicyType.java
@@ -22,7 +22,7 @@ public enum AccessPolicyType {
    /**
     * public
     */
-    PUBLIC("1", BucketCannedACL.PUBLIC_READ_WRITE, ObjectCannedACL.PUBLIC_READ_WRITE),
+    PUBLIC("1", BucketCannedACL.PUBLIC_READ, ObjectCannedACL.PUBLIC_READ),

    /**
     * custom