diff --git a/yudao-framework/yudao-spring-boot-starter-mq/pom.xml b/yudao-framework/yudao-spring-boot-starter-mq/pom.xml index c8972f1..8a677fd 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/pom.xml +++ b/yudao-framework/yudao-spring-boot-starter-mq/pom.xml @@ -38,6 +38,18 @@ rocketmq-spring-boot-starter true + + + + org.mockito + mockito-inline + test + + + org.springframework.boot + spring-boot-starter-test + test + \ No newline at end of file diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java index d0b7ae3..d8709ca 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java @@ -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> listeners; private final RedisMQTemplate redisTemplate; private final RedissonClient redissonClient; @@ -74,8 +81,9 @@ public class RedisPendingMessageResendJob { Map 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> 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; } // 重新投递消息 diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/test/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJobTest.java b/yudao-framework/yudao-spring-boot-starter-mq/src/test/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJobTest.java new file mode 100644 index 0000000..d005368 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/test/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJobTest.java @@ -0,0 +1,175 @@ +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 redisTemplate; + @Mock + private StreamOperations 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 body = singletonMap("payload", "{\"id\":1}"); + MapRecord 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> 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 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 buildIdRange(RecordId recordId) { + return Range.of(Range.Bound.inclusive(recordId.getValue()), Range.Bound.inclusive(recordId.getValue())); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java index 31c5078..48a7da3 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java @@ -64,7 +64,12 @@ public class IotDeviceMessageSubscriber implements IotMessageSubscriber { // 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()); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java index 4ea7338..d91d3ed 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceImpl.java @@ -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 sceneRules = getMatchedSceneRuleListByMessage(message); diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriberTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriberTest.java new file mode 100644 index 0000000..96df722 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriberTest.java @@ -0,0 +1,93 @@ +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()); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java index c337483..36496b2 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/test/java/cn/iocoder/yudao/module/iot/service/rule/scene/IotSceneRuleServiceSimpleTest.java @@ -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 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); + } + }