fix: 完善自动导入异常中断与恢复

- 自动导入基础设施异常触发批次熔断，保留完整日志并输出安全错误信息 - 增加恢复令牌与租约围栏、无文档失败项重建及消息退避机制 - 前端展示中断状态并在状态请求失败后自动恢复轮询 - 补充批次中断迁移、配置与并发异常路径测试
2026-08-10 11:26:00 +08:00
parent 54d85ae460
commit bae7b18977
33 changed files with 3725 additions and 221 deletions
--- a/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/config/MQProperties.java
+++ b/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/config/MQProperties.java
@@ -44,6 +44,9 @@ public class MQProperties {
        private int consumerBatchSize = 200;
        private Duration consumerBlockTimeout = Duration.ofMillis(2000);
        private Duration pendingClaimIdle = Duration.ofMillis(60000);
+        private Duration consumerFailureInitialBackoff = Duration.ofSeconds(1);
+        private Duration consumerFailureMaxBackoff = Duration.ofSeconds(30);
+        private Duration consumerFailureLogInterval = Duration.ofMinutes(1);
        private int maxRetry = 16;
        private ConsumerExecutor consumerExecutor = new ConsumerExecutor();
        private Pool pool = new Pool();
@@ -116,6 +119,63 @@ public class MQProperties {
            this.pendingClaimIdle = pendingClaimIdle;
        }

+        /**
+         * 获取消费循环首次失败退避时间。
+         *
+         * @return 首次失败退避时间
+         */
+        public Duration getConsumerFailureInitialBackoff() {
+            return consumerFailureInitialBackoff;
+        }
+
+        /**
+         * 设置消费循环首次失败退避时间。
+         *
+         * @param consumerFailureInitialBackoff 首次失败退避时间
+         */
+        public void setConsumerFailureInitialBackoff(
+            Duration consumerFailureInitialBackoff) {
+            this.consumerFailureInitialBackoff = consumerFailureInitialBackoff;
+        }
+
+        /**
+         * 获取消费循环最大失败退避时间。
+         *
+         * @return 最大失败退避时间
+         */
+        public Duration getConsumerFailureMaxBackoff() {
+            return consumerFailureMaxBackoff;
+        }
+
+        /**
+         * 设置消费循环最大失败退避时间。
+         *
+         * @param consumerFailureMaxBackoff 最大失败退避时间
+         */
+        public void setConsumerFailureMaxBackoff(
+            Duration consumerFailureMaxBackoff) {
+            this.consumerFailureMaxBackoff = consumerFailureMaxBackoff;
+        }
+
+        /**
+         * 获取消费循环完整异常日志间隔。
+         *
+         * @return 完整异常日志间隔
+         */
+        public Duration getConsumerFailureLogInterval() {
+            return consumerFailureLogInterval;
+        }
+
+        /**
+         * 设置消费循环完整异常日志间隔。
+         *
+         * @param consumerFailureLogInterval 完整异常日志间隔
+         */
+        public void setConsumerFailureLogInterval(
+            Duration consumerFailureLogInterval) {
+            this.consumerFailureLogInterval = consumerFailureLogInterval;
+        }
+
        public int getMaxRetry() {
            return maxRetry;
        }
--- a/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/core/MQDeferException.java
+++ b/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/core/MQDeferException.java
@@ -0,0 +1,21 @@
+package tech.easyflow.common.mq.core;
+
+/**
+ * 请求 MQ 暂缓确认当前消息。
+ *
+ * <p>消费者遇到暂时无法持久化终态的基础设施故障时抛出该异常。
+ * Redis Stream 容器会保留消息 pending 状态，等待超过 claim idle 后
+ * 再次领取，避免立即复制消息形成重试风暴。</p>
+ */
+public class MQDeferException extends RuntimeException {
+
+    /**
+     * 创建暂缓确认异常。
+     *
+     * @param message 暂缓原因
+     * @param cause 原始异常
+     */
+    public MQDeferException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}
--- a/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/redis/RedisMQConsumerContainer.java
+++ b/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/redis/RedisMQConsumerContainer.java
@@ -20,6 +20,7 @@ import tech.easyflow.common.mq.core.MQAcknowledger;
 import tech.easyflow.common.mq.core.MQConsumerContainer;
 import tech.easyflow.common.mq.core.MQConsumerHandler;
 import tech.easyflow.common.mq.core.MQDeadLetterService;
+import tech.easyflow.common.mq.core.MQDeferException;
 import tech.easyflow.common.mq.core.MQMessage;
 import tech.easyflow.common.mq.core.MQMessageConverter;
 import tech.easyflow.common.mq.core.MQSubscription;
@@ -157,11 +158,20 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
    private void consumeLoop(MQConsumerHandler handler, MQSubscription subscription, int shard) {
        String streamKey = keySupport.streamKey(subscription.getTopic(), shard);
        String consumerName = buildConsumerName(subscription.getConsumerGroup(), shard);
-        ensureConsumerGroup(streamKey, subscription.getConsumerGroup());
-        LOG.info("MQ 消费循环已启动: topic={}, group={}, shard={}, consumer={}, streamKey={}, handler={}",
-            subscription.getTopic(), subscription.getConsumerGroup(), shard, consumerName, streamKey, handler.getClass().getSimpleName());
+        int consecutiveFailures = 0;
+        long nextFullErrorLogAt = 0L;
+        int suppressedFailureLogs = 0;
+        boolean consumerGroupReady = false;
+        LOG.info(
+            "MQ 消费循环已启动: topic={}, group={}, shard={}, consumer={}, streamKey={}, handler={}",
+            subscription.getTopic(), subscription.getConsumerGroup(), shard,
+            consumerName, streamKey, handler.getClass().getSimpleName());
        while (running) {
            try {
+                if (!consumerGroupReady) {
+                    ensureConsumerGroup(streamKey, subscription.getConsumerGroup());
+                    consumerGroupReady = true;
+                }
                List<MapRecord<String, Object, Object>> pendingRecords =
                        reclaimPending(streamKey, subscription.getConsumerGroup(), consumerName);
                if (!pendingRecords.isEmpty()) {
@@ -171,6 +181,12 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                            subscription.getTopic(), subscription.getConsumerGroup(), shard, consumerName,
                            streamKey, pendingMessages.size());
                        handleMessages(handler, subscription, streamKey, subscription.getConsumerGroup(), pendingMessages);
+                        logConsumerRecovery(
+                            subscription, shard, consumerName, streamKey,
+                            handler, consecutiveFailures, suppressedFailureLogs);
+                        consecutiveFailures = 0;
+                        nextFullErrorLogAt = 0L;
+                        suppressedFailureLogs = 0;
                        continue;
                    }
                }
@@ -182,6 +198,12 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                        StreamOffset.create(streamKey, org.springframework.data.redis.connection.stream.ReadOffset.lastConsumed())
                );
                if (records == null || records.isEmpty()) {
+                    logConsumerRecovery(
+                        subscription, shard, consumerName, streamKey,
+                        handler, consecutiveFailures, suppressedFailureLogs);
+                    consecutiveFailures = 0;
+                    nextFullErrorLogAt = 0L;
+                    suppressedFailureLogs = 0;
                    continue;
                }
                List<MQMessage> messages = toMessages(streamKey, records);
@@ -191,20 +213,113 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                LOG.info("MQ 收到消息批次: topic={}, group={}, shard={}, consumer={}, streamKey={}, count={}",
                    subscription.getTopic(), subscription.getConsumerGroup(), shard, consumerName, streamKey, messages.size());
                handleMessages(handler, subscription, streamKey, subscription.getConsumerGroup(), messages);
+                logConsumerRecovery(
+                    subscription, shard, consumerName, streamKey,
+                    handler, consecutiveFailures, suppressedFailureLogs);
+                consecutiveFailures = 0;
+                nextFullErrorLogAt = 0L;
+                suppressedFailureLogs = 0;
            } catch (Exception exception) {
-                LOG.error("MQ 消费循环异常: topic={}, group={}, shard={}, consumer={}, streamKey={}, handler={}",
-                    subscription.getTopic(),
-                    subscription.getConsumerGroup(),
-                    shard,
-                    consumerName,
-                    streamKey,
-                    handler.getClass().getSimpleName(),
-                    exception);
-                sleepSilently(1000L);
+                consumerGroupReady = false;
+                consecutiveFailures++;
+                long now = System.currentTimeMillis();
+                if (now >= nextFullErrorLogAt) {
+                    LOG.error(
+                        "MQ 消费循环异常: topic={}, group={}, shard={}, consumer={}, "
+                            + "streamKey={}, handler={}, consecutiveFailures={}, "
+                            + "suppressedFailureLogs={}",
+                        subscription.getTopic(),
+                        subscription.getConsumerGroup(),
+                        shard,
+                        consumerName,
+                        streamKey,
+                        handler.getClass().getSimpleName(),
+                        consecutiveFailures,
+                        suppressedFailureLogs,
+                        exception);
+                    nextFullErrorLogAt = now + positiveMillis(
+                        properties.getRedis().getConsumerFailureLogInterval(),
+                        Duration.ofMinutes(1));
+                    suppressedFailureLogs = 0;
+                } else {
+                    suppressedFailureLogs++;
+                }
+                sleepSilently(calculateConsumerFailureBackoffMillis(
+                    consecutiveFailures));
            }
        }
    }

+    /**
+     * 计算消费循环连续失败后的指数退避时间。
+     *
+     * @param consecutiveFailures 连续失败次数
+     * @return 退避毫秒数
+     */
+    long calculateConsumerFailureBackoffMillis(int consecutiveFailures) {
+        long initialMillis = positiveMillis(
+            properties.getRedis().getConsumerFailureInitialBackoff(),
+            Duration.ofSeconds(1));
+        long maxMillis = Math.max(
+            initialMillis,
+            positiveMillis(
+                properties.getRedis().getConsumerFailureMaxBackoff(),
+                Duration.ofSeconds(30)));
+        int shift = Math.min(Math.max(0, consecutiveFailures - 1), 20);
+        long multiplier = 1L << shift;
+        if (initialMillis > maxMillis / multiplier) {
+            return maxMillis;
+        }
+        return Math.min(maxMillis, initialMillis * multiplier);
+    }
+
+    /**
+     * 在消费循环恢复后输出一次汇总日志。
+     *
+     * @param subscription 消费订阅
+     * @param shard 分片
+     * @param consumerName 消费者名称
+     * @param streamKey Stream Key
+     * @param handler 消费处理器
+     * @param consecutiveFailures 连续失败次数
+     * @param suppressedFailureLogs 已抑制日志数
+     */
+    private void logConsumerRecovery(MQSubscription subscription,
+                                     int shard,
+                                     String consumerName,
+                                     String streamKey,
+                                     MQConsumerHandler handler,
+                                     int consecutiveFailures,
+                                     int suppressedFailureLogs) {
+        if (consecutiveFailures <= 0) {
+            return;
+        }
+        LOG.info(
+            "MQ 消费循环已恢复: topic={}, group={}, shard={}, consumer={}, "
+                + "streamKey={}, handler={}, previousFailures={}, "
+                + "suppressedFailureLogs={}",
+            subscription.getTopic(),
+            subscription.getConsumerGroup(),
+            shard,
+            consumerName,
+            streamKey,
+            handler.getClass().getSimpleName(),
+            consecutiveFailures,
+            suppressedFailureLogs);
+    }
+
+    /**
+     * 将空值或非正数时长替换为安全默认值。
+     *
+     * @param duration 配置时长
+     * @param fallback 默认时长
+     * @return 正数毫秒值
+     */
+    private long positiveMillis(Duration duration, Duration fallback) {
+        long millis = duration == null ? 0L : duration.toMillis();
+        return millis > 0L ? millis : fallback.toMillis();
+    }
+
    /**
     * 构建 Redis Stream consumer name。
     *
@@ -314,6 +429,14 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                group, streamKey, messages.size(), handler.getClass().getSimpleName());
            return;
        } catch (Exception batchEx) {
+            if (isDeferred(batchEx)) {
+                LOG.warn(
+                    "MQ 批量消息暂缓确认，等待 pending 重领: group={}, "
+                        + "streamKey={}, count={}, handler={}, reason={}",
+                    group, streamKey, messages.size(),
+                    handler.getClass().getSimpleName(), resolveReason(batchEx));
+                throw batchEx;
+            }
            LOG.error("MQ 批量处理消息失败，准备降级单条处理: group={}, streamKey={}, count={}, handler={}",
                group, streamKey, messages.size(), handler.getClass().getSimpleName(), batchEx);
            if (messages.size() == 1) {
@@ -329,8 +452,10 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
    private void handleMessagesIndividually(MQConsumerHandler handler,
                                            String streamKey,
                                            String group,
-                                            List<MQMessage> messages) {
+                                            List<MQMessage> messages)
+        throws Exception {
        for (MQMessage message : messages) {
+            boolean shouldAcknowledge = true;
            try {
                LOG.info("MQ 开始单条处理消息: group={}, streamKey={}, messageId={}, handler={}",
                    group, streamKey, message.getMessageId(), handler.getClass().getSimpleName());
@@ -338,13 +463,42 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                LOG.info("MQ 单条处理消息完成: group={}, streamKey={}, messageId={}, handler={}",
                    group, streamKey, message.getMessageId(), handler.getClass().getSimpleName());
            } catch (Exception singleEx) {
-                retryOrDeadLetter(List.of(message), resolveReason(singleEx));
-            } finally {
+                if (isDeferred(singleEx)) {
+                    shouldAcknowledge = false;
+                    LOG.warn(
+                        "MQ 消息暂缓确认，停止当前批次并等待 pending 重领: "
+                            + "group={}, streamKey={}, messageId={}, handler={}, reason={}",
+                        group, streamKey, message.getMessageId(),
+                        handler.getClass().getSimpleName(),
+                        resolveReason(singleEx));
+                    throw singleEx;
+                } else {
+                    retryOrDeadLetter(List.of(message), resolveReason(singleEx));
+                }
+            }
+            if (shouldAcknowledge) {
                acknowledge(streamKey, group, List.of(message));
            }
        }
    }

+    /**
+     * 判断异常链是否请求暂缓消息确认。
+     *
+     * @param error 消费异常
+     * @return 是否暂缓确认
+     */
+    private boolean isDeferred(Throwable error) {
+        Throwable current = error;
+        while (current != null) {
+            if (current instanceof MQDeferException) {
+                return true;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
+
    private void acknowledge(String streamKey, String group, List<MQMessage> messages) {
        if (messages == null || messages.isEmpty()) {
            return;
--- a/easyflow-commons/easyflow-common-mq/src/test/java/tech/easyflow/common/mq/redis/RedisMQConsumerContainerTest.java
+++ b/easyflow-commons/easyflow-common-mq/src/test/java/tech/easyflow/common/mq/redis/RedisMQConsumerContainerTest.java
@@ -17,6 +17,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import tech.easyflow.common.mq.config.MQProperties;
 import tech.easyflow.common.mq.core.MQConsumerHandler;
 import tech.easyflow.common.mq.core.MQDeadLetterService;
+import tech.easyflow.common.mq.core.MQDeferException;
 import tech.easyflow.common.mq.core.MQMessage;
 import tech.easyflow.common.mq.core.MQMessageConverter;
 import tech.easyflow.common.mq.core.MQSubscription;
@@ -78,6 +79,85 @@ public class RedisMQConsumerContainerTest {
        Mockito.verify(streamOperations).acknowledge("stream-1", "group-1", "2-0");
    }

+    /**
+     * 验证暂时性基础设施故障保留当前及后续消息 pending，避免立即复制重投。
+     *
+     * @throws Exception 消息处理异常
+     */
+    @Test
+    public void handleMessagesShouldLeavePendingWhenConsumerRequestsDeferral()
+        throws Exception {
+        StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked")
+        StreamOperations<String, Object, Object> streamOperations =
+            Mockito.mock(StreamOperations.class);
+        Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations);
+        MQConsumerHandler handler = new MQConsumerHandler() {
+            @Override
+            public MQSubscription subscription() {
+                return new MQSubscription();
+            }
+
+            @Override
+            public void handle(List<MQMessage> messages) {
+                throw new MQDeferException(
+                    "数据库暂不可用", new IllegalStateException("database down"));
+            }
+        };
+        MQSubscription subscription = new MQSubscription();
+        subscription.setBatchEnabled(false);
+        RedisMQConsumerContainer container = container(redisTemplate, null);
+        MQMessage first = message("message-1", "1-0");
+        MQMessage second = message("message-2", "2-0");
+
+        try {
+            container.handleMessages(
+                handler, subscription, "stream-1", "group-1",
+                List.of(first, second));
+            Assert.fail("暂缓确认必须传播到消费循环以触发退避");
+        } catch (MQDeferException expected) {
+            Assert.assertEquals("数据库暂不可用", expected.getMessage());
+        }
+
+        Mockito.verify(streamOperations, Mockito.never()).acknowledge(
+            ArgumentMatchers.anyString(),
+            ArgumentMatchers.anyString(),
+            ArgumentMatchers.any(String[].class));
+        Mockito.verify(streamOperations, Mockito.never()).add(
+            ArgumentMatchers.any(MapRecord.class));
+    }
+
+    /**
+     * 验证消费循环失败退避按指数增长并受最大值限制。
+     */
+    @Test
+    public void consumerFailureBackoffShouldGrowExponentiallyWithCap() {
+        StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class);
+        MQProperties properties = new MQProperties();
+        properties.getRedis().setConsumerFailureInitialBackoff(
+            Duration.ofSeconds(1));
+        properties.getRedis().setConsumerFailureMaxBackoff(
+            Duration.ofSeconds(8));
+        RedisMQConsumerContainer container = new RedisMQConsumerContainer(
+            null,
+            redisTemplate,
+            properties,
+            new PlainMessageConverter(),
+            Mockito.mock(MQDeadLetterService.class),
+            null,
+            List.of()
+        );
+
+        Assert.assertEquals(1_000L,
+            container.calculateConsumerFailureBackoffMillis(1));
+        Assert.assertEquals(2_000L,
+            container.calculateConsumerFailureBackoffMillis(2));
+        Assert.assertEquals(8_000L,
+            container.calculateConsumerFailureBackoffMillis(4));
+        Assert.assertEquals(8_000L,
+            container.calculateConsumerFailureBackoffMillis(12));
+    }
+
    /**
     * 验证 pending 消息被 claim 后可以转换为 MQ 消息继续消费。
     */