fix(iot): 修复删除设备后消息反复重投及重投任务扫描无界

YunaiV
1 parent 8701cf75
Showing 7 changed files with 342 additions and 9 deletions
yudao-framework/yudao-spring-boot-starter-mq/pom.xml
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java
yudao-framework/yudao-spring-boot-starter-mq/src/test/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJobTest.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriberTest.java
yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java
@@ -38,6 +38,18 @@
             <artifactId>rocketmq-spring-boot-starter</artifactId>
             <optional>true</optional>
         </dependency>
+
+        <!-- Test 测试相关 -->
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-inline</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
 \ No newline at end of file
@@ -35,6 +35,13 @@ public class RedisPendingMessageResendJob {
      */
     private static final int EXPIRE_TIME = 5 * 60;
+    /**
+     * 每个消费者每轮最多扫描的 pending 消息数量
+     *
+     * 避免 pending 堆积时，单次 XPENDING 返回过多数据而长时间占用 Redis；未扫描到的消息，由后续轮次继续处理
+     */
+    private static final int PENDING_BATCH_SIZE = 1000;
+
     private final List<AbstractRedisStreamMessageListener<?>> listeners;
     private final RedisMQTemplate redisTemplate;
     private final RedissonClient redissonClient;
@@ -74,8 +81,9 @@ public class RedisPendingMessageResendJob {
             Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
             pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> {
                 log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
-                // 每个消费者的 pending消息的详情信息
-                PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(listener.getGroup(), consumerName), Range.unbounded(), pendingMessageCount);
+                // 每个消费者的 pending消息的详情信息，分批扫描：https://github.com/YunaiV/ruoyi-vue-pro/issues/1271
+                PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(listener.getGroup(), consumerName),
+                        Range.unbounded(), Math.min(pendingMessageCount, PENDING_BATCH_SIZE));
                 if (pendingMessages.isEmpty()) {
                     return;
                 }
@@ -89,6 +97,11 @@ public class RedisPendingMessageResendJob {
                     List<MapRecord<String, Object, Object>> records = ops.range(listener.getStreamKey(),
                             Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString())));
                     if (CollUtil.isEmpty(records)) {
+                        // 消息体已不存在（例如被 RedisStreamMessageCleanupJob 裁剪），无法重新投递，直接 ack 避免 pending 永久滞留
+                        redisTemplate.getRedisTemplate().opsForStream().acknowledge(listener.getStreamKey(),
+                                listener.getGroup(), pendingMessage.getId());
+                        log.warn("[processPendingMessage][消息({}) 的消息体已不存在，无法重新投递，直接 ack]",
+                                pendingMessage.getIdAsString());
                         return;
                     }
                     // 重新投递消息
+package cn.iocoder.yudao.framework.mq.redis.core.job;
+
+import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate;
+import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessage;
+import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.dao.QueryTimeoutException;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.stream.Record;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StreamOperations;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link RedisPendingMessageResendJob} 的单元测试类
+ *
+ * @author 芋道源码
+ */
+@ExtendWith(MockitoExtension.class)
+public class RedisPendingMessageResendJobTest {
+
+    private static final String STREAM_KEY = "test_stream";
+    private static final String GROUP = "test_group";
+    private static final String CONSUMER_NAME = "test_consumer";
+
+    private RedisPendingMessageResendJob resendJob;
+
+    @Mock
+    private RedisMQTemplate redisMQTemplate;
+    @Mock
+    private RedisTemplate<String, Object> redisTemplate;
+    @Mock
+    private StreamOperations<String, Object, Object> streamOperations;
+    @Mock
+    private RedissonClient redissonClient;
+    @Mock
+    private RLock lock;
+
+    @BeforeEach
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    public void setUp() {
+        AbstractRedisStreamMessageListener<?> listener = new AbstractRedisStreamMessageListener<>(STREAM_KEY, GROUP) {
+
+            @Override
+            public void onMessage(AbstractRedisStreamMessage message) {
+            }
+
+        };
+        resendJob = new RedisPendingMessageResendJob(singletonList(listener), redisMQTemplate, redissonClient,
+                RedisPendingMessageResendJob.DEFAULT_RESEND_LOCK_KEY);
+        // mock 分布式锁
+        when(redissonClient.getLock(eq(RedisPendingMessageResendJob.DEFAULT_RESEND_LOCK_KEY))).thenReturn(lock);
+        when(lock.tryLock()).thenReturn(true);
+        when(lock.isHeldByCurrentThread()).thenReturn(true);
+        // mock Redis Stream 操作
+        when(redisMQTemplate.getRedisTemplate()).thenReturn((RedisTemplate) redisTemplate);
+        when(redisTemplate.opsForStream()).thenReturn((StreamOperations) streamOperations);
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testMessageResend_success() {
+        // mock 数据：一条超时消息，一条未超时消息
+        RecordId expiredId = RecordId.of("1-0");
+        RecordId freshId = RecordId.of("2-0");
+        mockPending(2L, List.of(buildPendingMessage(expiredId, Duration.ofMinutes(6)),
+                buildPendingMessage(freshId, Duration.ofMinutes(1))));
+        Map<Object, Object> body = singletonMap("payload", "{\"id\":1}");
+        MapRecord<String, Object, Object> record = StreamRecords.newRecord().in(STREAM_KEY).withId(expiredId).ofMap(body);
+        when(streamOperations.range(eq(STREAM_KEY), eq(buildIdRange(expiredId)))).thenReturn(singletonList(record));
+
+        // 调用
+        resendJob.messageResend();
+        // 断言：超时消息重新投递完整消息体，并 ack 原 pending 消息
+        ArgumentCaptor<Record<String, ?>> recordCaptor = ArgumentCaptor.forClass(Record.class);
+        verify(streamOperations).add(recordCaptor.capture());
+        assertEquals(STREAM_KEY, recordCaptor.getValue().getStream());
+        assertEquals(body, recordCaptor.getValue().getValue());
+        verify(streamOperations).acknowledge(eq(GROUP), eq(record));
+        // 断言：未超时消息不处理
+        verify(streamOperations, never()).range(eq(STREAM_KEY), eq(buildIdRange(freshId)));
+        verify(lock).unlock();
+    }
+
+    @Test
+    public void testMessageResend_messageBodyNotExists() {
+        // mock 数据：消息体已被裁剪
+        RecordId recordId = RecordId.of("1-0");
+        mockPending(1L, singletonList(buildPendingMessage(recordId, Duration.ofMinutes(6))));
+        when(streamOperations.range(eq(STREAM_KEY), eq(buildIdRange(recordId)))).thenReturn(Collections.emptyList());
+
+        // 调用
+        resendJob.messageResend();
+        // 断言：无法重新投递，直接 ack 清理 pending
+        verify(streamOperations).acknowledge(eq(STREAM_KEY), eq(GROUP), eq(recordId));
+        verify(streamOperations, never()).add(any(Record.class));
+    }
+
+    @Test
+    public void testMessageResend_pendingBatchSize() {
+        // mock 数据：pending 数量超过单批上限
+        mockPending(5000L, Collections.emptyList());
+
+        // 调用
+        resendJob.messageResend();
+        // 断言：单次 XPENDING 最多扫描 1000 条
+        verify(streamOperations).pending(eq(STREAM_KEY), eq(Consumer.from(GROUP, CONSUMER_NAME)),
+                eq(Range.unbounded()), eq(1000L));
+    }
+
+    @Test
+    public void testMessageResend_pendingLessThanBatchSize() {
+        // mock 数据：pending 数量未超过单批上限
+        mockPending(3L, Collections.emptyList());
+
+        // 调用
+        resendJob.messageResend();
+        // 断言：按实际数量扫描
+        verify(streamOperations).pending(eq(STREAM_KEY), eq(Consumer.from(GROUP, CONSUMER_NAME)),
+                eq(Range.unbounded()), eq(3L));
+    }
+
+    @Test
+    public void testMessageResend_redisException() {
+        // mock 数据：读取消息体时 Redis 超时
+        RecordId recordId = RecordId.of("1-0");
+        mockPending(1L, singletonList(buildPendingMessage(recordId, Duration.ofMinutes(6))));
+        when(streamOperations.range(eq(STREAM_KEY), eq(buildIdRange(recordId))))
+                .thenThrow(new QueryTimeoutException("Redis 超时"));
+
+        // 调用
+        resendJob.messageResend();
+        // 断言：保留 pending，不 ack 也不重新投递，等待下一轮重试
+        verify(streamOperations, never()).acknowledge(anyString(), anyString(), any(RecordId[].class));
+        verify(streamOperations, never()).acknowledge(anyString(), any(Record.class));
+        verify(streamOperations, never()).add(any(Record.class));
+        verify(lock).unlock();
+    }
+
+    // ========== 随机对象 ==========
+
+    private void mockPending(Long pendingCount, List<PendingMessage> pendingMessages) {
+        when(streamOperations.pending(eq(STREAM_KEY), eq(GROUP))).thenReturn(new PendingMessagesSummary(GROUP,
+                pendingCount, Range.unbounded(), singletonMap(CONSUMER_NAME, pendingCount)));
+        when(streamOperations.pending(eq(STREAM_KEY), eq(Consumer.from(GROUP, CONSUMER_NAME)), eq(Range.unbounded()),
+                anyLong())).thenReturn(new PendingMessages(GROUP, pendingMessages));
+    }
+
+    private static PendingMessage buildPendingMessage(RecordId recordId, Duration elapsedTime) {
+        return new PendingMessage(recordId, Consumer.from(GROUP, CONSUMER_NAME), elapsedTime, 1L);
+    }
+
+    private static Range<String> buildIdRange(RecordId recordId) {
+        return Range.of(Range.Bound.inclusive(recordId.getValue()), Range.Bound.inclusive(recordId.getValue()));
+    }
+
+}
@@ -64,7 +64,12 @@ public class IotDeviceMessageSubscriber implements IotMessageSubscriber&lt;IotDevic
         TenantUtils.execute(message.getTenantId(), () -> {
             // 1.1 更新设备的最后时间
-            IotDeviceDO device = deviceService.validateDeviceExistsFromCache(message.getDeviceId());
+            // 设备已删除时（例如删除时消息仍在途），跳过处理，避免消息不断重投：https://github.com/YunaiV/ruoyi-vue-pro/issues/1271
+            IotDeviceDO device = deviceService.getDeviceFromCache(message.getDeviceId());
+            if (device == null) {
+                log.warn("[onMessage][message({}) 对应的设备不存在，忽略处理]", message);
+                return;
+            }
             devicePropertyService.updateDeviceReportTimeAsync(device.getId(), LocalDateTime.now());
             // 1.2 更新设备的连接 server
             devicePropertyService.updateDeviceServerIdAsync(device.getId(), message.getServerId());
@@ -200,6 +200,10 @@ public class IotSceneRuleServiceImpl implements IotSceneRuleService {
     public void executeSceneRuleByDevice(IotDeviceMessage message) {
         // 1.1 这里的 tenantId，通过设备获取；
         IotDeviceDO device = deviceService.getDeviceFromCache(message.getDeviceId());
+        if (device == null) { // 设备已删除时，跳过处理：https://github.com/YunaiV/ruoyi-vue-pro/issues/1271
+            log.warn("[executeSceneRuleByDevice][设备({}) 不存在]", message.getDeviceId());
+            return;
+        }
         TenantUtils.execute(device.getTenantId(), () -> {
             // 1.2 获得设备匹配的规则场景
             List<IotSceneRuleDO> sceneRules = getMatchedSceneRuleListByMessage(message);
+package cn.iocoder.yudao.module.iot.mq.consumer.device;
+
+import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.device.IotDeviceStateEnum;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDeviceDO;
+import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
+import cn.iocoder.yudao.module.iot.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.service.device.property.IotDevicePropertyService;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.springframework.dao.QueryTimeoutException;
+
+import java.time.LocalDateTime;
+
+import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomLongId;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link IotDeviceMessageSubscriber} 的单元测试类
+ *
+ * @author 芋道源码
+ */
+public class IotDeviceMessageSubscriberTest extends BaseMockitoUnitTest {
+
+    @InjectMocks
+    private IotDeviceMessageSubscriber deviceMessageSubscriber;
+
+    @Mock
+    private IotDeviceService deviceService;
+    @Mock
+    private IotDevicePropertyService devicePropertyService;
+    @Mock
+    private IotDeviceMessageService deviceMessageService;
+    @Mock
+    private IotMessageBus messageBus;
+
+    @Test
+    public void testOnMessage_success() {
+        // 准备参数
+        IotDeviceMessage message = buildPropertyPostMessage();
+        // mock 方法：设备在线
+        IotDeviceDO device = new IotDeviceDO().setId(message.getDeviceId())
+                .setState(IotDeviceStateEnum.ONLINE.getState());
+        when(deviceService.getDeviceFromCache(eq(message.getDeviceId()))).thenReturn(device);
+
+        // 调用
+        deviceMessageSubscriber.onMessage(message);
+        // 断言
+        verify(devicePropertyService).updateDeviceReportTimeAsync(eq(device.getId()), any(LocalDateTime.class));
+        verify(devicePropertyService).updateDeviceServerIdAsync(eq(device.getId()), eq(message.getServerId()));
+        verify(deviceMessageService).handleUpstreamDeviceMessage(eq(message), eq(device));
+    }
+
+    @Test
+    public void testOnMessage_deviceNotExists() {
+        // 准备参数：设备已删除时，消息仍在途
+        IotDeviceMessage message = buildPropertyPostMessage();
+        // mock 方法
+        when(deviceService.getDeviceFromCache(eq(message.getDeviceId()))).thenReturn(null);
+
+        // 调用：正常返回，由消息总线 ack，不再重投
+        assertDoesNotThrow(() -> deviceMessageSubscriber.onMessage(message));
+        // 断言
+        verifyNoInteractions(devicePropertyService, deviceMessageService);
+    }
+
+    @Test
+    public void testOnMessage_queryException() {
+        // 准备参数
+        IotDeviceMessage message = buildPropertyPostMessage();
+        // mock 方法：查询设备时 Redis 超时
+        when(deviceService.getDeviceFromCache(eq(message.getDeviceId())))
+                .thenThrow(new QueryTimeoutException("Redis 超时"));
+
+        // 调用，并断言异常：可恢复异常继续抛出，消息不 ack，由重投任务重试
+        assertThrows(QueryTimeoutException.class, () -> deviceMessageSubscriber.onMessage(message));
+        verifyNoInteractions(devicePropertyService, deviceMessageService);
+    }
+
+    private static IotDeviceMessage buildPropertyPostMessage() {
+        return new IotDeviceMessage().setDeviceId(randomLongId()).setTenantId(randomLongId())
+                .setServerId("test-server").setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+    }
+
+}
@@ -3,6 +3,8 @@ package cn.iocoder.yudao.module.iot.service.rule.scene;
 import cn.iocoder.yudao.framework.common.enums.CommonStatusEnum;
 import cn.iocoder.yudao.framework.test.core.ut.BaseMockitoUnitTest;
 import cn.iocoder.yudao.module.iot.controller.admin.rule.vo.scene.IotSceneRuleSaveReqVO;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotSceneRuleDO;
 import cn.iocoder.yudao.module.iot.dal.mysql.rule.IotSceneRuleMapper;
 import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
@@ -14,6 +16,7 @@ import cn.iocoder.yudao.module.iot.service.rule.scene.timer.IotTimerConditionEva
 import org.junit.jupiter.api.Test;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
+import org.springframework.dao.QueryTimeoutException;
 import java.util.Collections;
 import java.util.List;
@@ -22,6 +25,7 @@ import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomLongId
 import static cn.iocoder.yudao.framework.test.core.util.RandomUtils.randomPojo;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
@@ -38,22 +42,16 @@ public class IotSceneRuleServiceSimpleTest extends BaseMockitoUnitTest {
     @Mock
     private IotSceneRuleMapper sceneRuleMapper;
-
     @Mock
     private List<IotSceneRuleAction> sceneRuleActions;
-
     @Mock
     private IotSceneRuleTimerHandler timerHandler;
-
     @Mock
     private IotTimerConditionEvaluator timerConditionEvaluator;
-
     @Mock
     private IotSceneRuleMatcherManager sceneRuleMatcherManager;
-
     @Mock
     private IotProductService productService;
-
     @Mock
     private IotDeviceService deviceService;
@@ -216,4 +214,37 @@ public class IotSceneRuleServiceSimpleTest extends BaseMockitoUnitTest {
         // 验证
         verify(sceneRuleMapper, times(1)).selectById(id);
     }
+
+    @Test
+    public void testExecuteSceneRuleByDevice_deviceNotExists() {
+        // 准备参数：设备已删除时，消息仍在途
+        IotDeviceMessage message = new IotDeviceMessage().setDeviceId(randomLongId())
+                .setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+
+        // Mock 行为
+        when(deviceService.getDeviceFromCache(message.getDeviceId())).thenReturn(null);
+
+        // 调用 - 设备不存在时跳过，不抛出异常
+        assertDoesNotThrow(() -> sceneRuleService.executeSceneRuleByDevice(message));
+
+        // 验证：不再匹配、执行场景规则
+        verify(deviceService, times(1)).getDeviceFromCache(message.getDeviceId());
+        verifyNoInteractions(sceneRuleMapper, productService, sceneRuleMatcherManager, sceneRuleActions);
+    }
+
+    @Test
+    public void testExecuteSceneRuleByDevice_queryException() {
+        // 准备参数
+        IotDeviceMessage message = new IotDeviceMessage().setDeviceId(randomLongId())
+                .setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+
+        // Mock 行为：查询设备时 Redis 超时
+        when(deviceService.getDeviceFromCache(message.getDeviceId()))
+                .thenThrow(new QueryTimeoutException("Redis 超时"));
+
+        // 调用，并断言异常：可恢复异常继续抛出，由消息重投重试
+        assertThrows(QueryTimeoutException.class, () -> sceneRuleService.executeSceneRuleByDevice(message));
+        verifyNoInteractions(sceneRuleMapper, productService, sceneRuleMatcherManager, sceneRuleActions);
+    }
+
 }