feat: 增强多实例分布式部署兼容

- 增加定时任务分布式锁并覆盖 chatlog、文档导入和 Agent HITL 过期扫描 - 增强 Redis MQ 多实例 consumer 标识、pending reclaim 和单条处理能力 - 增加文档导入状态 Redis 广播和 Agent HITL 跨节点路由确认
2026-05-29 18:27:46 +08:00
parent cc3bb9cff0
commit 0f4d10c43c
39 changed files with 2703 additions and 17 deletions
--- a/easyflow-commons/easyflow-common-mq/pom.xml
+++ b/easyflow-commons/easyflow-common-mq/pom.xml
@@ -27,5 +27,17 @@
            <artifactId>commons-pool2</artifactId>
            <version>2.11.1</version>
        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>5.12.0</version>
+            <scope>test</scope>
+        </dependency>
    </dependencies>
 </project>
--- 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
@@ -1,9 +1,13 @@
 package tech.easyflow.common.mq.config;

 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.util.StringUtils;

 import java.time.Duration;

+/**
+ * EasyFlow MQ 配置。
+ */
@ConfigurationProperties(prefix = "easyflow.mq")
 public class MQProperties {

@@ -35,6 +39,7 @@ public class MQProperties {

        private int database = 1;
        private String streamPrefix = "easyflow:mq";
+        private String consumerInstanceId = defaultConsumerInstanceId();
        private int chatPersistShardCount = 4;
        private int consumerBatchSize = 200;
        private Duration consumerBlockTimeout = Duration.ofMillis(2000);
@@ -59,6 +64,26 @@ public class MQProperties {
            this.streamPrefix = streamPrefix;
        }

+        /**
+         * 获取 Redis Stream 消费实例 ID。
+         *
+         * @return 消费实例 ID
+         */
+        public String getConsumerInstanceId() {
+            return consumerInstanceId;
+        }
+
+        /**
+         * 设置 Redis Stream 消费实例 ID。
+         *
+         * @param consumerInstanceId 消费实例 ID
+         */
+        public void setConsumerInstanceId(String consumerInstanceId) {
+            this.consumerInstanceId = StringUtils.hasText(consumerInstanceId)
+                ? consumerInstanceId.trim()
+                : defaultConsumerInstanceId();
+        }
+
        public int getChatPersistShardCount() {
            return chatPersistShardCount;
        }
@@ -191,5 +216,13 @@ public class MQProperties {
                this.minIdle = minIdle;
            }
        }
+
+        private static String defaultConsumerInstanceId() {
+            String hostName = System.getenv("HOSTNAME");
+            if (StringUtils.hasText(hostName)) {
+                return hostName.trim();
+            }
+            return java.util.UUID.randomUUID().toString();
+        }
    }
 }
--- a/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/core/MQSubscription.java
+++ b/easyflow-commons/easyflow-common-mq/src/main/java/tech/easyflow/common/mq/core/MQSubscription.java
@@ -5,6 +5,7 @@ public class MQSubscription {
    private String topic;
    private String consumerGroup;
    private int shardCount;
+    private boolean batchEnabled = true;

    public String getTopic() {
        return topic;
@@ -29,4 +30,22 @@ public class MQSubscription {
    public void setShardCount(int shardCount) {
        this.shardCount = shardCount;
    }
+
+    /**
+     * 是否启用批量消费。
+     *
+     * @return true 表示启用批量消费
+     */
+    public boolean isBatchEnabled() {
+        return batchEnabled;
+    }
+
+    /**
+     * 设置是否启用批量消费。
+     *
+     * @param batchEnabled 是否启用批量消费
+     */
+    public void setBatchEnabled(boolean batchEnabled) {
+        this.batchEnabled = batchEnabled;
+    }
 }
--- 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
@@ -30,6 +30,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.regex.Pattern;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -39,6 +40,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifecycle {

    private static final Logger LOG = LoggerFactory.getLogger(RedisMQConsumerContainer.class);
+    private static final Pattern UNSAFE_CONSUMER_NAME_CHARS = Pattern.compile("[^A-Za-z0-9_.-]");

    private final RedisConnectionFactory redisConnectionFactory;
    private final StringRedisTemplate stringRedisTemplate;
@@ -154,13 +156,24 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec

    private void consumeLoop(MQConsumerHandler handler, MQSubscription subscription, int shard) {
        String streamKey = keySupport.streamKey(subscription.getTopic(), shard);
-        String consumerName = subscription.getConsumerGroup() + "-" + 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());
        while (running) {
            try {
-                reclaimPending(streamKey, subscription.getConsumerGroup(), consumerName);
+                List<MapRecord<String, Object, Object>> pendingRecords =
+                        reclaimPending(streamKey, subscription.getConsumerGroup(), consumerName);
+                if (!pendingRecords.isEmpty()) {
+                    List<MQMessage> pendingMessages = toMessages(streamKey, pendingRecords);
+                    if (!pendingMessages.isEmpty()) {
+                        LOG.info("MQ 收到重领 pending 消息批次: topic={}, group={}, shard={}, consumer={}, streamKey={}, count={}",
+                            subscription.getTopic(), subscription.getConsumerGroup(), shard, consumerName,
+                            streamKey, pendingMessages.size());
+                        handleMessages(handler, subscription, streamKey, subscription.getConsumerGroup(), pendingMessages);
+                        continue;
+                    }
+                }
                List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(
                        Consumer.from(subscription.getConsumerGroup(), consumerName),
                        StreamReadOptions.empty()
@@ -177,7 +190,7 @@ 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, streamKey, subscription.getConsumerGroup(), messages);
+                handleMessages(handler, subscription, streamKey, subscription.getConsumerGroup(), messages);
            } catch (Exception exception) {
                LOG.error("MQ 消费循环异常: topic={}, group={}, shard={}, consumer={}, streamKey={}, handler={}",
                    subscription.getTopic(),
@@ -192,7 +205,20 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
        }
    }

-    private void reclaimPending(String streamKey, String group, String consumerName) {
+    /**
+     * 构建 Redis Stream consumer name。
+     *
+     * @param consumerGroup 消费组
+     * @param shard 分片序号
+     * @return consumer name
+     */
+    String buildConsumerName(String consumerGroup, int shard) {
+        String instanceId = properties.getRedis().getConsumerInstanceId();
+        String safeInstanceId = UNSAFE_CONSUMER_NAME_CHARS.matcher(instanceId).replaceAll("-");
+        return consumerGroup + "-" + shard + "-" + safeInstanceId;
+    }
+
+    List<MapRecord<String, Object, Object>> reclaimPending(String streamKey, String group, String consumerName) {
        Duration idle = properties.getRedis().getPendingClaimIdle();
        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
            RedisStreamCommands.XPendingOptions options = RedisStreamCommands.XPendingOptions
@@ -200,7 +226,7 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
            var pendingMessages = connection.streamCommands()
                    .xPending(streamKey.getBytes(StandardCharsets.UTF_8), group, options);
            if (pendingMessages == null || pendingMessages.isEmpty()) {
-                return;
+                return List.of();
            }
            List<RecordId> ids = new ArrayList<>();
            for (PendingMessage pendingMessage : pendingMessages) {
@@ -209,15 +235,16 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
                }
            }
            if (ids.isEmpty()) {
-                return;
+                return List.of();
            }
-            stringRedisTemplate.opsForStream().claim(
+            List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().claim(
                    streamKey,
                    group,
                    consumerName,
                    idle,
                    ids.toArray(new RecordId[0])
            );
+            return records == null ? List.of() : records;
        }
    }

@@ -233,7 +260,7 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
        }
    }

-    private List<MQMessage> toMessages(String streamKey, List<MapRecord<String, Object, Object>> records) {
+    List<MQMessage> toMessages(String streamKey, List<MapRecord<String, Object, Object>> records) {
        List<MQMessage> messages = new ArrayList<>(records.size());
        for (MapRecord<String, Object, Object> record : records) {
            Object payload = record.getValue().get("payload");
@@ -269,7 +296,15 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
        }
    }

-    private void handleMessages(MQConsumerHandler handler, String streamKey, String group, List<MQMessage> messages) throws Exception {
+    void handleMessages(MQConsumerHandler handler,
+                        MQSubscription subscription,
+                        String streamKey,
+                        String group,
+                        List<MQMessage> messages) throws Exception {
+        if (!subscription.isBatchEnabled()) {
+            handleMessagesIndividually(handler, streamKey, group, messages);
+            return;
+        }
        try {
            LOG.info("MQ 开始批量处理消息: group={}, streamKey={}, count={}, handler={}",
                group, streamKey, messages.size(), handler.getClass().getSimpleName());
@@ -288,6 +323,13 @@ public class RedisMQConsumerContainer implements MQConsumerContainer, SmartLifec
            }
        }

+        handleMessagesIndividually(handler, streamKey, group, messages);
+    }
+
+    private void handleMessagesIndividually(MQConsumerHandler handler,
+                                            String streamKey,
+                                            String group,
+                                            List<MQMessage> messages) {
        for (MQMessage message : messages) {
            try {
                LOG.info("MQ 开始单条处理消息: group={}, streamKey={}, messageId={}, handler={}",
--- 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
@@ -0,0 +1,175 @@
+package tech.easyflow.common.mq.redis;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisStreamCommands;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessage;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.core.StreamOperations;
+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.MQMessage;
+import tech.easyflow.common.mq.core.MQMessageConverter;
+import tech.easyflow.common.mq.core.MQSubscription;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * {@link RedisMQConsumerContainer} 回归测试。
+ */
+public class RedisMQConsumerContainerTest {
+
+    /**
+     * 验证 consumer name 包含稳定实例 ID，且消费组名称不被改变。
+     */
+    @Test
+    public void buildConsumerNameShouldAppendSanitizedInstanceId() {
+        MQProperties properties = new MQProperties();
+        properties.getRedis().setConsumerInstanceId("node/a:1");
+        RedisMQConsumerContainer container = new RedisMQConsumerContainer(
+            null,
+            null,
+            properties,
+            null,
+            null,
+            null,
+            List.of()
+        );
+
+        String consumerName = container.buildConsumerName("chat-persist", 2);
+
+        Assert.assertEquals("chat-persist-2-node-a-1", consumerName);
+    }
+
+    /**
+     * 验证关闭批量消费后，容器按单条处理并独立确认消息。
+     *
+     * @throws Exception 消息处理异常
+     */
+    @Test
+    public void handleMessagesShouldProcessIndividuallyWhenBatchDisabled() 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);
+        RecordingHandler handler = new RecordingHandler();
+        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");
+
+        container.handleMessages(handler, subscription, "stream-1", "group-1", List.of(first, second));
+
+        Assert.assertEquals(List.of(List.of("message-1"), List.of("message-2")), handler.calls);
+        Mockito.verify(streamOperations).acknowledge("stream-1", "group-1", "1-0");
+        Mockito.verify(streamOperations).acknowledge("stream-1", "group-1", "2-0");
+    }
+
+    /**
+     * 验证 pending 消息被 claim 后可以转换为 MQ 消息继续消费。
+     */
+    @Test
+    public void reclaimPendingShouldReturnClaimedRecordsForConsumption() {
+        StringRedisTemplate redisTemplate = Mockito.mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked")
+        StreamOperations<String, Object, Object> streamOperations = Mockito.mock(StreamOperations.class);
+        Mockito.when(redisTemplate.opsForStream()).thenReturn(streamOperations);
+        RedisConnectionFactory connectionFactory = Mockito.mock(RedisConnectionFactory.class);
+        RedisConnection connection = Mockito.mock(RedisConnection.class);
+        RedisStreamCommands streamCommands = Mockito.mock(RedisStreamCommands.class);
+        Mockito.when(connectionFactory.getConnection()).thenReturn(connection);
+        Mockito.when(connection.streamCommands()).thenReturn(streamCommands);
+        PendingMessage pendingMessage = new PendingMessage(
+                RecordId.of("1-0"), Consumer.from("group-1", "old-consumer"), Duration.ofMinutes(2), 1);
+        Mockito.when(streamCommands.xPending(
+                        ArgumentMatchers.eq("stream-1".getBytes(java.nio.charset.StandardCharsets.UTF_8)),
+                        ArgumentMatchers.eq("group-1"),
+                        ArgumentMatchers.any(RedisStreamCommands.XPendingOptions.class)))
+                .thenReturn(new PendingMessages("group-1", List.of(pendingMessage)));
+        Map<Object, Object> payload = Map.of("payload", "message-1");
+        MapRecord<String, Object, Object> record = MapRecord
+                .create("stream-1", payload)
+                .withId(RecordId.of("1-0"));
+        Mockito.when(streamOperations.claim(
+                        ArgumentMatchers.eq("stream-1"),
+                        ArgumentMatchers.eq("group-1"),
+                        ArgumentMatchers.eq("consumer-1"),
+                        ArgumentMatchers.any(Duration.class),
+                        ArgumentMatchers.any(RecordId[].class)))
+                .thenReturn(List.of(record));
+        RedisMQConsumerContainer container = container(redisTemplate, connectionFactory);
+
+        List<MapRecord<String, Object, Object>> records =
+                container.reclaimPending("stream-1", "group-1", "consumer-1");
+        List<MQMessage> messages = container.toMessages("stream-1", records);
+
+        Assert.assertEquals(1, records.size());
+        Assert.assertEquals(1, messages.size());
+        Assert.assertEquals("message-1", messages.get(0).getMessageId());
+        Assert.assertEquals("1-0", messages.get(0).getStreamMessageId());
+    }
+
+    private RedisMQConsumerContainer container(StringRedisTemplate redisTemplate,
+                                              RedisConnectionFactory connectionFactory) {
+        MQProperties properties = new MQProperties();
+        return new RedisMQConsumerContainer(
+                connectionFactory,
+                redisTemplate,
+                properties,
+                new PlainMessageConverter(),
+                Mockito.mock(MQDeadLetterService.class),
+                null,
+                List.of()
+        );
+    }
+
+    private MQMessage message(String messageId, String streamMessageId) {
+        MQMessage message = new MQMessage();
+        message.setMessageId(messageId);
+        message.setStreamMessageId(streamMessageId);
+        return message;
+    }
+
+    private static final class RecordingHandler implements MQConsumerHandler {
+
+        private final List<List<String>> calls = new ArrayList<>();
+
+        @Override
+        public MQSubscription subscription() {
+            return new MQSubscription();
+        }
+
+        @Override
+        public void handle(List<MQMessage> messages) {
+            calls.add(messages.stream().map(MQMessage::getMessageId).toList());
+        }
+    }
+
+    private static final class PlainMessageConverter implements MQMessageConverter {
+
+        @Override
+        public String serialize(MQMessage message) {
+            return message.getMessageId();
+        }
+
+        @Override
+        public MQMessage deserialize(String payload) {
+            MQMessage message = new MQMessage();
+            message.setMessageId(payload);
+            return message;
+        }
+    }
+}