feat(mq-starter-redis-stream): 订阅 stream 定时裁剪，复用 redis-stream.trim.* 默认开启

底层中间件的 trim 任务只认自己注册的 handler（无 handler 不启动），mq-starter 不走 handler 体系，此前只经 mq-starter 收发的 topic 无界涨 Redis 内存。driver 现自带 RedisStreamTrimScheduler：订阅时登记 stream，按 trim.interval-ms（默认 60s）对登记过的 stream 执行 XTRIM MAXLEN ~ maxlen（默认 10 万），stop 时随 driver 停止。 - maxlen<=0 视为关闭（XTRIM 0 会清空整条 stream，绝不能发出） - 只裁订阅过的 stream，纯发送方不裁，与中间件 handler 即消费者的语义对齐 - 单 stream 裁剪失败仅 WARN，不中断其余 stream - README trim/autoclaim 行改述真实行为，CLAUDE.md 模块描述同步注意：默认开启，升级后订阅 stream 开始按 maxlen 封顶；裁剪不看消费进度，堆积超限时最老的未消费消息会被删，容量按可容忍停机时长×生产速率估算。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-16 10:25:29 +01:00
parent 2b635a4cd7
commit 8bba71236f
5 changed files with 264 additions and 4 deletions
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java
@@ -27,6 +27,7 @@ public class RedisStreamMqDriver implements MqDriver {
    private final RedisStreamProperties props;
    private final long reclaimMinIdleMs;
    private final List<Thread> workers = new ArrayList<>();
+    private final RedisStreamTrimScheduler trimScheduler;
    private volatile boolean running = true;

    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
@@ -38,6 +39,7 @@ public class RedisStreamMqDriver implements MqDriver {
    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
        this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
        this.reclaimMinIdleMs = reclaimMinIdleMs;
+        this.trimScheduler = new RedisStreamTrimScheduler(redis, props);
    }

    @Override public void send(String topic, MqMessage message) {
@@ -57,6 +59,7 @@ public class RedisStreamMqDriver implements MqDriver {
        String stream = keyBuilder.buildStream(sub.getTopic());
        String group = sub.getGroup();
        ensureGroup(stream, group);
+        trimScheduler.register(stream);
        String consumer = (props.getConsumerName() == null || props.getConsumerName().isEmpty())
                ? "mq-" + UUID.randomUUID() : props.getConsumerName();
        Thread t = new Thread(() -> loop(sub, stream, group, consumer), "mq-redis-" + stream + "-" + group);
@@ -205,5 +208,5 @@ public class RedisStreamMqDriver implements MqDriver {
    }
    private static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } }

-    @Override public void stop() { running = false; for (Thread t : workers) t.interrupt(); workers.clear(); }
+    @Override public void stop() { running = false; trimScheduler.stop(); for (Thread t : workers) t.interrupt(); workers.clear(); }
 }
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamTrimScheduler.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamTrimScheduler.java
@@ -0,0 +1,76 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 订阅 stream 的定时裁剪（{@code XTRIM <stream> MAXLEN ~ <maxlen>}，近似裁剪）。
+ *
+ * <p>底层中间件的 StreamTrimScheduler 只裁自己注册的 handler 的 stream，而 mq-starter 的消费
+ * 不走 handler 体系——driver 不自己裁的话，只经 mq-starter 收发的 topic 无界增长。配置复用中间件的
+ * {@code redis-stream.trim.*}（默认开启：maxlen 10 万 / 周期 60s）。
+ *
+ * <p>两条语义约束：{@code maxlen<=0} 视为关闭——XTRIM 0 会清空整条 stream，绝不能发出；
+ * 只裁「订阅过」的 stream——与中间件“handler 即消费者”的语义对齐，纯发送方不裁
+ * （裁剪节奏该由消费方掌握，发送方不知道还有谁没消费到哪）。
+ */
+@Slf4j
+class RedisStreamTrimScheduler {
+    private final StringRedisTemplate redis;
+    private final RedisStreamProperties props;
+    private final Set<String> streams = ConcurrentHashMap.newKeySet();
+    private ScheduledExecutorService scheduler;
+
+    RedisStreamTrimScheduler(StringRedisTemplate redis, RedisStreamProperties props) {
+        this.redis = redis;
+        this.props = props;
+    }
+
+    /** 订阅时登记 stream；首个登记且 maxlen&gt;0 时拉起调度线程（重复登记幂等）。 */
+    synchronized void register(String stream) {
+        streams.add(stream);
+        long maxlen = props.getTrim().getMaxlen();
+        if (scheduler != null || maxlen <= 0) return;
+        long interval = props.getTrim().getIntervalMs();
+        scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mq-redis-trim");
+            t.setDaemon(true);
+            return t;
+        });
+        scheduler.scheduleWithFixedDelay(this::trimAll, interval, interval, TimeUnit.MILLISECONDS);
+        log.info("[mq-redis] stream 裁剪调度启动 interval={}ms maxlen={}（近似裁剪，只裁订阅过的 stream）", interval, maxlen);
+    }
+
+    void trimAll() {
+        long maxlen = props.getTrim().getMaxlen();
+        if (maxlen <= 0) return;
+        for (String stream : streams) {
+            try {
+                redis.execute((RedisCallback<Object>) c -> c.execute("XTRIM",
+                        stream.getBytes(StandardCharsets.UTF_8),
+                        "MAXLEN".getBytes(StandardCharsets.UTF_8),
+                        "~".getBytes(StandardCharsets.UTF_8),
+                        String.valueOf(maxlen).getBytes(StandardCharsets.UTF_8)));
+            } catch (Exception e) {
+                // 单条失败不得中断其余 stream 的裁剪
+                log.warn("[mq-redis] stream 裁剪失败 stream={}", stream, e);
+            }
+        }
+    }
+
+    synchronized void stop() {
+        if (scheduler != null) {
+            scheduler.shutdownNow();
+            scheduler = null;
+        }
+    }
+}
--- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTrimTest.java
+++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTrimTest.java
@@ -0,0 +1,181 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.mq.spi.MqSubscription;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StreamOperations;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * 订阅 stream 的定时裁剪（XTRIM MAXLEN ~）。
+ *
+ * <p>底层中间件的 trim 任务只认自己注册的 handler，mq-starter 不走那套——
+ * 所以 driver 必须自己裁订阅过的 stream，否则只经 mq-starter 收发的 topic 无界增长。
+ * 配置复用中间件的 {@code redis-stream.trim.*}（maxlen 默认 10 万 / 周期默认 60s），默认开启。
+ */
+class RedisStreamMqDriverTrimTest {
+
+    /** 记录 connection.execute 收到的每次调用：[command, arg1, arg2, ...]（args 按 UTF-8 还原成字符串）。 */
+    private final List<List<String>> executed = Collections.synchronizedList(new ArrayList<>());
+    private StringRedisTemplate redis;
+
+    @BeforeEach
+    void mockRedis() {
+        redis = mock(StringRedisTemplate.class);
+        RedisConnection conn = recordingConnection(null);
+        when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
+                ((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
+    }
+
+    /**
+     * 记录 execute 调用的 RedisConnection——用默认 Answer 而非参数匹配，
+     * 因为 execute(String, byte[]...) 是 varargs，这版 Mockito 的 any(byte[][].class) 匹配不上。
+     */
+    private RedisConnection recordingConnection(String failStream) {
+        return mock(RedisConnection.class, inv -> {
+            if ("toString".equals(inv.getMethod().getName())) return "mock-conn";
+            if (!"execute".equals(inv.getMethod().getName())) return null;
+            List<String> call = new ArrayList<>();
+            for (Object arg : inv.getArguments()) {
+                if (arg instanceof byte[]) call.add(new String((byte[]) arg, StandardCharsets.UTF_8));
+                else if (arg instanceof byte[][]) {
+                    for (byte[] b : (byte[][]) arg) call.add(new String(b, StandardCharsets.UTF_8));
+                } else call.add(String.valueOf(arg));
+            }
+            if (failStream != null && call.size() > 1 && failStream.equals(call.get(1))) {
+                throw new IllegalStateException("boom");
+            }
+            executed.add(call);
+            return null;
+        });
+    }
+
+    private static RedisStreamProperties propsWithTrim(long maxlen, long intervalMs) {
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.getTrim().setMaxlen(maxlen);
+        props.getTrim().setIntervalMs(intervalMs);
+        props.setConsumerName("test-consumer");
+        return props;
+    }
+
+    private List<List<String>> xtrims() {
+        synchronized (executed) {
+            return executed.stream().filter(c -> "XTRIM".equals(c.get(0))).collect(Collectors.toList());
+        }
+    }
+
+    // ---- RedisStreamTrimScheduler 单元 ----
+
+    @Test
+    void trimAll_issuesApproximateXtrimForEachRegisteredStream() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(50000, 60000));
+        trim.register("order_dev");
+        trim.register("pay_dev");
+
+        trim.trimAll();
+
+        List<List<String>> calls = xtrims();
+        assertThat(calls).hasSize(2);
+        assertThat(calls).allSatisfy(c -> {
+            assertThat(c.subList(2, 5)).as("XTRIM <stream> MAXLEN ~ <maxlen>")
+                    .containsExactly("MAXLEN", "~", "50000");
+        });
+        assertThat(calls.stream().map(c -> c.get(1)))
+                .containsExactlyInAnyOrder("order_dev", "pay_dev");
+        trim.stop();
+    }
+
+    @Test
+    void register_sameStreamTwice_trimsOnlyOnce() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
+        trim.register("order_dev");
+        trim.register("order_dev");
+
+        trim.trimAll();
+
+        assertThat(xtrims()).hasSize(1);
+        trim.stop();
+    }
+
+    @Test
+    void maxlenNotPositive_treatedAsDisabled_neverTrims() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(0, 60000));
+        trim.register("order_dev");
+
+        trim.trimAll();
+
+        assertThat(xtrims()).as("maxlen<=0 视为关闭，XTRIM 0 会清空整条 stream，绝不能发出").isEmpty();
+        trim.stop();
+    }
+
+    @Test
+    void oneStreamFails_othersStillTrimmed() {
+        RedisConnection conn = recordingConnection("bad_dev");
+        when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
+                ((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
+
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
+        trim.register("bad_dev");
+        trim.register("good_dev");
+
+        trim.trimAll(); // bad_dev 抛异常不得中断 good_dev
+
+        List<List<String>> calls = xtrims();
+        assertThat(calls).hasSize(1);
+        assertThat(calls.get(0).get(1)).isEqualTo("good_dev");
+        trim.stop();
+    }
+
+    // ---- driver 接线：订阅登记 + 周期触发 + 停机停止 ----
+
+    @Test
+    void subscribe_startsPeriodicTrimOfSubscribedStream_andStopHaltsIt() throws Exception {
+        RedisStreamProperties props = propsWithTrim(1234, 50); // 50ms 一裁，便于观察
+        StreamOperations<String, Object, Object> streamOps = mock(StreamOperations.class);
+        when(redis.opsForStream()).thenReturn(streamOps);
+        when(streamOps.read(any(org.springframework.data.redis.connection.stream.Consumer.class),
+                any(), any())).thenAnswer(inv -> { Thread.sleep(20); return Collections.emptyList(); });
+
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(
+                redis, new StreamKeyBuilder(true, "dev"), props);
+        driver.subscribe(MqSubscription.builder()
+                .topic("t-order").group("g1").tag("*").maxRetry(3)
+                .payloadType(String.class).listener(msg -> { }).build());
+
+        List<List<String>> calls = awaitXtrim(2000);
+        assertThat(calls).as("订阅后应周期性裁剪").isNotEmpty();
+        assertThat(calls.get(0).get(1)).as("裁的是拼了环境后缀的实际 stream 名").isEqualTo("t-order_dev");
+        assertThat(calls.get(0).subList(2, 5)).containsExactly("MAXLEN", "~", "1234");
+
+        driver.stop();
+        Thread.sleep(120); // 让在途 tick 收尾
+        int settled = xtrims().size();
+        Thread.sleep(250);
+        assertThat(xtrims()).as("stop 后不得再裁").hasSize(settled);
+    }
+
+    private List<List<String>> awaitXtrim(long timeoutMs) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            List<List<String>> calls = xtrims();
+            if (!calls.isEmpty()) return calls;
+            Thread.sleep(20);
+        }
+        return xtrims();
+    }
+}