浇水bug修改

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttClientManager.java

@@ -37,65 +37,18 @@ public class MqttClientManager implements DisposableBean {
     private final MqttMessageDispatcher dispatcher;
     private MqttAsyncClient client;
     private ThreadPoolExecutor consumerExecutor;
-    private BlockingQueue<InboundMessage> messageQueue;
+    private List<BlockingQueue<InboundMessage>> messageQueues;
     private volatile boolean running;
+    private int consumerCount;
 
-    @Bean
-    @ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
-    public MqttAsyncClient mqttConnect() throws MqttException {
-        if (StringUtils.isBlank(props.getBrokerUrl())) {
-            throw new ServiceException("启用 MQTT 时必须配置 broker-url");
+    static int deviceIdentityHash(String topic) {
+        if (topic == null) {
+            return 0;
         }
-        if (StringUtils.isBlank(props.getClientId())) {
-            throw new ServiceException("启用 MQTT 时必须配置 client-id");
-        }
-        messageQueue = new ArrayBlockingQueue<>(Math.max(1, props.getAsync().getQueueCapacity()));
-        consumerExecutor = createConsumerExecutor();
-        running = true;
-        startConsumers();
-        client = new MqttAsyncClient(props.getBrokerUrl(), props.getClientId(), new MemoryPersistence());
-
-        MqttConnectOptions options = new MqttConnectOptions();
-        options.setUserName(props.getUsername());
-        if (props.getPassword() != null) {
-            options.setPassword(props.getPassword().toCharArray());
-        }
-        options.setKeepAliveInterval(props.getKeepAlive());
-        options.setConnectionTimeout(props.getConnectionTimeout());
-        options.setAutomaticReconnect(props.isAutomaticReconnect());
-        options.setCleanSession(props.isCleanSession());
-        options.setMaxInflight(props.getMaxInflight());
-        configureTls(options);
-
-        client.setCallback(new MqttCallbackExtended() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.warn("[MQTT] 连接已断开：{}", cause == null ? "" : cause.getMessage());
-            }
-
-            @Override
-            public void connectComplete(boolean reconnect, String serverURI) {
-                if (reconnect) {
-                    log.info("[MQTT] 已重新连接：{}", serverURI);
-                    subscribeTopics();
-                }
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) {
-                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
-                enqueue(topic, payload);
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-            }
-        });
-
-        client.connect(options).waitForCompletion();
-        log.info("[MQTT] 已连接：{}", props.getBrokerUrl());
-        subscribeTopics();
-        return client;
+        int start = topic.startsWith("/") ? 1 : 0;
+        int end = topic.indexOf('/', start);
+        String identity = end < 0 ? topic.substring(start) : topic.substring(start, end);
+        return identity.hashCode();
     }
 
     private void configureTls(MqttConnectOptions options) throws MqttException {
@@ -171,39 +124,97 @@ public class MqttClientManager implements DisposableBean {
         };
     }
 
+    @Bean
+    @ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
+    public MqttAsyncClient mqttConnect() throws MqttException {
+        if (StringUtils.isBlank(props.getBrokerUrl())) {
+            throw new ServiceException("启用 MQTT 时必须配置 broker-url");
+        }
+        if (StringUtils.isBlank(props.getClientId())) {
+            throw new ServiceException("启用 MQTT 时必须配置 client-id");
+        }
+        consumerCount = Math.max(1, props.getAsync().getConsumerCount());
+        messageQueues = createMessageQueues();
+        consumerExecutor = createConsumerExecutor();
+        running = true;
+        startConsumers();
+        client = new MqttAsyncClient(props.getBrokerUrl(), props.getClientId(), new MemoryPersistence());
+
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(props.getUsername());
+        if (props.getPassword() != null) {
+            options.setPassword(props.getPassword().toCharArray());
+        }
+        options.setKeepAliveInterval(props.getKeepAlive());
+        options.setConnectionTimeout(props.getConnectionTimeout());
+        options.setAutomaticReconnect(props.isAutomaticReconnect());
+        options.setCleanSession(props.isCleanSession());
+        options.setMaxInflight(props.getMaxInflight());
+        configureTls(options);
+
+        client.setCallback(new MqttCallbackExtended() {
+            @Override
+            public void connectionLost(Throwable cause) {
+                log.warn("[MQTT] 连接已断开：{}", cause == null ? "" : cause.getMessage());
+            }
+
+            @Override
+            public void connectComplete(boolean reconnect, String serverURI) {
+                if (reconnect) {
+                    log.info("[MQTT] 已重新连接：{}", serverURI);
+                    subscribeTopics();
+                }
+            }
+
+            @Override
+            public void messageArrived(String topic, MqttMessage message) {
+                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
+                enqueue(topic, payload, message.isRetained());
+            }
+
+            @Override
+            public void deliveryComplete(IMqttDeliveryToken token) {
+            }
+        });
+
+        client.connect(options).waitForCompletion();
+        log.info("[MQTT] 已连接：{}", props.getBrokerUrl());
+        subscribeTopics();
+        return client;
+    }
+
     private void startConsumers() {
-        int consumerCount = Math.max(1, props.getAsync().getConsumerCount());
         for (int i = 0; i < consumerCount; i++) {
-            consumerExecutor.execute(this::consumeLoop);
+            final int consumerIndex = i;
+            consumerExecutor.execute(() -> consumeLoop(consumerIndex));
         }
     }
 
-    private void enqueue(String topic, String payload) {
-        InboundMessage inboundMessage = new InboundMessage(topic, payload);
+    private void enqueue(String topic, String payload, boolean retained) {
+        InboundMessage inboundMessage = new InboundMessage(topic, payload, retained);
         try {
-            if (!messageQueue.offer(inboundMessage, props.getAsync().getOfferTimeoutMs(), TimeUnit.MILLISECONDS)) {
-                log.warn("[MQTT] 上行消息队列已满，丢弃 主题={}", topic);
-            }
+            messageQueues.get(queueIndex(topic)).put(inboundMessage);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             log.warn("[MQTT] 上行消息入队被中断 主题={}", topic);
         }
     }
 
-    private void consumeLoop() {
+    private void consumeLoop(int consumerIndex) {
         MqttProperties.Async async = props.getAsync();
         int batchSize = Math.max(1, async.getBatchSize());
         long pollTimeoutMs = Math.max(1, async.getPollTimeoutMs());
         List<InboundMessage> batch = new java.util.ArrayList<>(batchSize);
+        BlockingQueue<InboundMessage> queue = messageQueues.get(consumerIndex);
 
-        while (running || !messageQueue.isEmpty()) {
+        while (running || !queue.isEmpty()) {
             try {
-                InboundMessage first = messageQueue.poll(pollTimeoutMs, TimeUnit.MILLISECONDS);
+                InboundMessage first = queue.poll(pollTimeoutMs, TimeUnit.MILLISECONDS);
                 if (first == null) {
                     continue;
                 }
                 batch.add(first);
-                messageQueue.drainTo(batch, batchSize - 1);
+                queue.drainTo(batch, batchSize - 1);
                 for (InboundMessage message : batch) {
                     dispatch(message);
                 }
@@ -218,12 +229,26 @@ public class MqttClientManager implements DisposableBean {
 
     private void dispatch(InboundMessage message) {
         try {
-            dispatcher.dispatch(message.topic(), message.payload());
+            dispatcher.dispatch(message.topic(), message.payload(), message.retained());
         } catch (Exception e) {
             log.error("[MQTT] 消息分发失败 主题={}", message.topic(), e);
         }
     }
 
+    private List<BlockingQueue<InboundMessage>> createMessageQueues() {
+        int capacity = Math.max(1, props.getAsync().getQueueCapacity());
+        int queueCapacity = Math.max(1, (capacity + Math.max(1, consumerCount) - 1) / Math.max(1, consumerCount));
+        List<BlockingQueue<InboundMessage>> queues = new java.util.ArrayList<>(consumerCount);
+        for (int i = 0; i < consumerCount; i++) {
+            queues.add(new ArrayBlockingQueue<>(queueCapacity));
+        }
+        return queues;
+    }
+
+    private int queueIndex(String topic) {
+        return Math.floorMod(deviceIdentityHash(topic), Math.max(1, consumerCount));
+    }
+
     private void subscribeTopics() {
         try {
             if (client == null || !client.isConnected()) {
@@ -276,6 +301,6 @@ public class MqttClientManager implements DisposableBean {
         }
     }
 
-    private record InboundMessage(String topic, String payload) {
+    private record InboundMessage(String topic, String payload, boolean retained) {
     }
 }

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

@@ -22,6 +22,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 
 @Slf4j
 @Service
@@ -43,11 +44,10 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
         pendingIds().add(command.getCommandId());
     }
 
-    public void removePending(String commandId) {
-        if (StringUtils.isBlank(commandId)) {
-            return;
-        }
-        withCommandLock(commandId, () -> deletePending(commandId));
+    static boolean ackMatchesPendingCommand(String topicDeviceNo, DeviceCommand pendingCommand) {
+        return StringUtils.isNotBlank(topicDeviceNo)
+            && pendingCommand != null
+            && topicDeviceNo.equals(pendingCommand.getDeviceNo());
     }
 
     static boolean resolveMissingCommandId(DeviceCommandAck ack, List<DeviceCommand> pendingCommands) {
@@ -68,6 +68,16 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
         return true;
     }
 
+    public void removePending(String commandId) {
+        if (StringUtils.isBlank(commandId)) {
+            return;
+        }
+        withCommandLock(commandId, () -> {
+            deletePending(commandId);
+            return true;
+        });
+    }
+
     @Override
     public void handleAck(String deviceNo, String payload) {
         if (StringUtils.isBlank(payload)) {
@@ -102,8 +112,19 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
             }
         }
         boolean ackSaved = withCommandLock(ack.getCommandId(), () -> {
+            DeviceCommand pending = RedisUtils.getCacheObject(pendingKey(ack.getCommandId()));
+            if (pending == null) {
+                log.warn("[MQTT] ACK 对应命令不存在或已过期 设备编号={} 命令编号={}", deviceNo, ack.getCommandId());
+                return false;
+            }
+            if (!ackMatchesPendingCommand(deviceNo, pending)) {
+                log.warn("[MQTT] ACK 设备编号与待确认命令不匹配，已拒绝 设备编号={} 命令编号={} 命令设备={}",
+                    deviceNo, ack.getCommandId(), pending.getDeviceNo());
+                return false;
+            }
             deletePending(ack.getCommandId());
             RedisUtils.setCacheObject(ackKey(ack.getCommandId()), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+            return true;
         });
         if (!ackSaved) {
             log.warn("[MQTT] ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, ack.getCommandId());
@@ -130,8 +151,19 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
         ack.setMessage(payload);
 
         boolean ackSaved = withCommandLock(commandId, () -> {
+            DeviceCommand pending = RedisUtils.getCacheObject(pendingKey(commandId));
+            if (pending == null) {
+                log.warn("[MQTT] 非 JSON ACK 对应命令不存在或已过期 设备编号={} 命令编号={}", deviceNo, commandId);
+                return false;
+            }
+            if (!ackMatchesPendingCommand(deviceNo, pending)) {
+                log.warn("[MQTT] 非 JSON ACK 设备编号与待确认命令不匹配，已拒绝 设备编号={} 命令编号={} 命令设备={}",
+                    deviceNo, commandId, pending.getDeviceNo());
+                return false;
+            }
             deletePending(commandId);
             RedisUtils.setCacheObject(ackKey(commandId), ack, Duration.ofSeconds(mqttProperties.getCommandAck().getAckTtlSeconds()));
+            return true;
         });
         if (!ackSaved) {
             log.warn("[MQTT] 非 JSON ACK 处理未获得命令锁，本次跳过 设备编号={} 命令编号={}", deviceNo, commandId);
@@ -160,7 +192,10 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
     }
 
     private void retryCommandIfLocked(String commandId, long now) {
-        withCommandLock(commandId, () -> retryCommand(commandId, now));
+        withCommandLock(commandId, () -> {
+            retryCommand(commandId, now);
+            return true;
+        });
     }
 
     private void retryCommand(String commandId, long now) {
@@ -178,24 +213,24 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
             return;
         }
         if (!isDeviceOnline(command.getDeviceNo())) {
-            command.setRetryCount(command.getRetryCount() + 1);
-            if (command.getRetryCount() >= mqttProperties.getCommandAck().getMaxRetryCount()) {
-                deletePending(commandId);
-                log.warn("[MQTT] 设备离线，命令达到最大等待次数，已删除待确认命令 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
-                return;
-            }
             command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
             savePending(command);
-            log.debug("[MQTT] 设备离线，命令延后重试 设备编号={} 命令编号={} 等待次数={}", command.getDeviceNo(), commandId, command.getRetryCount());
+            log.debug("[MQTT] 设备离线，命令延后重试 设备编号={} 命令编号={}", command.getDeviceNo(), commandId);
             return;
         }
 
         command.setRetryCount(command.getRetryCount() + 1);
         command.setLastSentAt(now);
         command.setNextRetryAt(now + mqttProperties.getCommandAck().getRetryIntervalMs());
-        mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
-        savePending(command);
-        log.info("[MQTT] 命令已重新下发 设备编号={} 命令编号={} 重试次数={}", command.getDeviceNo(), commandId, command.getRetryCount());
+        try {
+            mqttClientManager.publish(command.getTopic(), JsonUtils.toJsonString(command.getPayload()));
+            savePending(command);
+            log.info("[MQTT] 命令已重新下发 设备编号={} 命令编号={} 重试次数={}", command.getDeviceNo(), commandId, command.getRetryCount());
+        } catch (RuntimeException e) {
+            savePending(command);
+            log.warn("[MQTT] 命令重新下发失败，已延后重试 设备编号={} 命令编号={} 重试次数={}",
+                command.getDeviceNo(), commandId, command.getRetryCount(), e);
+        }
     }
 
     private boolean isDeviceOnline(String deviceNo) {
@@ -249,7 +284,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
         pendingIds().remove(commandId);
     }
 
-    private boolean withCommandLock(String commandId, Runnable action) {
+    private boolean withCommandLock(String commandId, BooleanSupplier action) {
         RLock lock = RedisUtils.getClient().getLock(mqttProperties.getCommandAck().getRetryLockKeyPrefix() + commandId);
         boolean locked = false;
         try {
@@ -257,8 +292,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler {
             if (!locked) {
                 return false;
             }
-            action.run();
-            return true;
+            return action.getAsBoolean();
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             log.warn("[MQTT] 命令锁等待被中断 命令编号={}", commandId);

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/config/properties/MqttProperties.java

@@ -18,7 +18,7 @@ public class MqttProperties {
     private String password;
     private String clientId;
     private int qos = 1;
-    private int keepAlive = 60;
+    private int keepAlive = 300;
     private int connectionTimeout = 30;
     private int maxInflight = 1000;
     private boolean cleanSession = false;
@@ -65,7 +65,7 @@ public class MqttProperties {
         private String retryLockKeyPrefix = "lock:mqtt:command:retry:";
         private long retryLockTtlMs = 30000;
         private String deviceStatusCachePrefix = "mqtt:device:status:";
-        private long deviceStatusCacheTtlSeconds = 300;
+        private long deviceStatusCacheTtlSeconds = 900;
     }
 
 }

water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/DeviceMqttCommandPublisherTest.java

@@ -0,0 +1,26 @@
+package org.dromara.mqtt;
+
+import org.dromara.mqtt.config.properties.MqttProperties;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.test.util.ReflectionTestUtils.invokeMethod;
+
+@Tag("dev")
+class DeviceMqttCommandPublisherTest {
+
+    @Test
+    void buildCommandTopicDefaultsToDeviceNo() {
+        MqttProperties properties = new MqttProperties();
+        DeviceMqttCommandPublisher publisher = new DeviceMqttCommandPublisher(
+            null,
+            null,
+            properties
+        );
+
+        String topic = invokeMethod(publisher, "buildCommandTopic", "D01");
+
+        assertThat(topic).isEqualTo("/d01/subscriber/cmd");
+    }
+}

water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java

@@ -0,0 +1,122 @@
+package org.dromara.mqtt;
+
+import cn.hutool.extra.spring.SpringUtil;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.apache.ibatis.session.Configuration;
+import org.dromara.app.domain.AppDevice;
+import org.dromara.app.domain.mqtt.DeviceCommand;
+import org.dromara.app.domain.mqtt.DeviceCommandAck;
+import org.dromara.app.mapper.AppDeviceMapper;
+import org.dromara.common.redis.utils.RedisUtils;
+import org.dromara.mqtt.config.properties.MqttProperties;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+
+@Tag("dev")
+class MqttCommandAckServiceTest {
+
+    private static GenericApplicationContext applicationContext;
+
+    @BeforeAll
+    static void initializeRedisUtils() {
+        if (TableInfoHelper.getTableInfo(AppDevice.class) == null) {
+            TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new Configuration(), ""), AppDevice.class);
+        }
+        applicationContext = new GenericApplicationContext();
+        applicationContext.registerBean(RedissonClient.class, () -> mock(RedissonClient.class));
+        applicationContext.refresh();
+        new SpringUtil().setApplicationContext(applicationContext);
+    }
+
+    @AfterAll
+    static void closeApplicationContext() {
+        applicationContext.close();
+    }
+
+    @Test
+    void resolveMissingCommandId_usesSinglePendingCommandForDevice() {
+        DeviceCommandAck ack = new DeviceCommandAck();
+        ack.setCommandId("");
+
+        DeviceCommand pending = new DeviceCommand();
+        pending.setCommandId("cmd-1");
+        pending.setDeviceNo("D01");
+
+        boolean resolved = MqttCommandAckService.resolveMissingCommandId(ack, List.of(pending));
+
+        assertThat(resolved).isTrue();
+        assertThat(ack.getCommandId()).isEqualTo("cmd-1");
+        assertThat(ack.getStatus()).isEqualTo("1");
+    }
+
+    @Test
+    void resolveMissingCommandId_refusesAmbiguousPendingCommands() {
+        DeviceCommandAck ack = new DeviceCommandAck();
+        ack.setCommandId("");
+
+        DeviceCommand first = new DeviceCommand();
+        first.setCommandId("cmd-1");
+        DeviceCommand second = new DeviceCommand();
+        second.setCommandId("cmd-2");
+
+        boolean resolved = MqttCommandAckService.resolveMissingCommandId(ack, List.of(first, second));
+
+        assertThat(resolved).isFalse();
+        assertThat(ack.getCommandId()).isEmpty();
+    }
+
+    @Test
+    void ackMatchesPendingCommandRequiresSameDevice() {
+        DeviceCommand pending = new DeviceCommand();
+        pending.setDeviceNo("D01");
+
+        assertThat(MqttCommandAckService.ackMatchesPendingCommand("D01", pending)).isTrue();
+        assertThat(MqttCommandAckService.ackMatchesPendingCommand("D02", pending)).isFalse();
+        assertThat(MqttCommandAckService.ackMatchesPendingCommand("D01", null)).isFalse();
+    }
+
+    @Test
+    void refreshDeviceOnlineRenewsStatusCacheTtl() throws Exception {
+        MqttProperties properties = new MqttProperties();
+        properties.getCommandAck().setDeviceStatusCacheTtlSeconds(900);
+        AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class);
+        RedissonClient redissonClient = mock(RedissonClient.class);
+        RLock lock = mock(RLock.class);
+        when(redissonClient.getLock("lock:mqtt:device:status:D01")).thenReturn(lock);
+        when(lock.tryLock(0, 10, TimeUnit.SECONDS)).thenReturn(true);
+        when(lock.isHeldByCurrentThread()).thenReturn(true);
+
+        MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper);
+        try (MockedStatic<RedisUtils> redis = mockStatic(RedisUtils.class)) {
+            redis.when(RedisUtils::getClient).thenReturn(redissonClient);
+
+            ReflectionTestUtils.invokeMethod(service, "refreshDeviceOnline", "D01");
+
+            redis.verify(() -> RedisUtils.setCacheObject(
+                eq("mqtt:device:status:D01"),
+                any(Map.class),
+                eq(Duration.ofSeconds(900))
+            ));
+            verify(appDeviceMapper).update(eq(null), any());
+            verify(lock).unlock();
+        }
+    }
+}