From 8bba71236f12e3584b53b1d943c111e82beba460 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 16 Jul 2026 10:25:29 +0100 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-redis-stream):=20=E8=AE=A2?= =?UTF-8?q?=E9=98=85=20stream=20=E5=AE=9A=E6=97=B6=E8=A3=81=E5=89=AA?= =?UTF-8?q?=EF=BC=8C=E5=A4=8D=E7=94=A8=20redis-stream.trim.*=20=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=E5=BC=80=E5=90=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 底层中间件的 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 --- CLAUDE.md | 2 +- README.md | 4 +- .../mq/driver/redis/RedisStreamMqDriver.java | 5 +- .../redis/RedisStreamTrimScheduler.java | 76 ++++++++ .../redis/RedisStreamMqDriverTrimTest.java | 181 ++++++++++++++++++ 5 files changed, 264 insertions(+), 4 deletions(-) create mode 100644 mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamTrimScheduler.java create mode 100644 mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTrimTest.java diff --git a/CLAUDE.md b/CLAUDE.md index 46d0f90..3e31953 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,7 +32,7 @@ mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest#someMethod # 单个 ``` mq-starter-core 门面 + 注解扫描 + 消费链 + SPI 契约（中立，不依赖 Spring/Redis 的核心编排逻辑，可纯单测） mq-starter-idempotent-redis 消费去重的 Redis 存储（driver 中立，不按 mq.type 门控；任何 driver 有 Redis 就能用） -mq-starter-redis-stream Redis Stream driver（去重/批量/落库/reclaim 重试/tag 消费端过滤全接入；去重 store 依赖 idempotent-redis 模块） +mq-starter-redis-stream Redis Stream driver（去重/批量/落库/reclaim 重试/tag 消费端过滤/订阅 stream 定时裁剪全接入；去重 store 依赖 idempotent-redis 模块。裁剪复用底层 redis-stream.trim.* 配置、默认开启，maxlen<=0 关，只裁订阅过的 stream——中间件自己的 trim 任务只认它的 handler，对 mq-starter 无效） mq-starter-rocketmq RocketMQ driver（收发 + 原生重试/死信 + RETRY_EXHAUSTED 落库 + 去重信号处理；去重 store 经聚合坐标间接获得） mq-spring-boot-starter 聚合 jar，业务方只引这一个（恒含 idempotent-redis） ``` diff --git a/README.md b/README.md index 24211cf..7125739 100644 --- a/README.md +++ b/README.md @@ -301,8 +301,8 @@ Redis 连接走标准 `spring.redis.*`。另有一组更底层的参数（前缀 | `block-ms` | 2000 | 拉取无消息时的阻塞等待毫秒数，到点空手返回进下一轮 | | `consumer-name` | 空 | 组内消费者名。留空则每次启动生成随机名（推荐，重启遗留的 PEL 由 reclaim 拿回）；写死会让多实例同名、PEL 归属混在一起 | | `max-retry` | 3 | 兜底重试次数，仅 `@MqListener` 的 `maxRetry` 配成 0/负数时才用到 | -| `trim.maxlen` / `trim.interval-ms` | 100000 / 60s | ⚠️ **对 mq-starter 消费的 stream 不生效**：中间件的定时裁剪只认自己注册的 handler 的 stream（无 handler 时任务不启动），mq-starter 不走那套 handler。只经 mq-starter 收发的 topic **没有任何裁剪、无界增长**，要封顶得靠直连中间件的服务碰巧订同一 topic，或运维层面自行 XTRIM | -| `autoclaim.*` | — | 同上对 mq-starter 不生效（也是按 handler 找 stream）。mq-starter 有自己的 reclaim，配上表的 `mq.redis-stream.reclaim-min-idle` | +| `trim.maxlen` / `trim.interval-ms` | 100000 / 60s | stream 封顶裁剪，**默认开启**：driver 对**订阅过**的 stream 每周期执行一次 `XTRIM MAXLEN ~ maxlen`（`~` 近似裁剪，实际长度可能略超 maxlen）。⚠️ 裁剪只按条数砍最老的、**不看消费进度**——堆积超过 maxlen 时未消费的老消息也会被删，容量按「可容忍的最大停机时长 × 生产速率」估算；`maxlen` 配 0/负数关闭；纯发送、不订阅的 topic 不裁（裁剪节奏归消费方） | +| `autoclaim.*` | — | 对 mq-starter **不生效**：中间件的 autoclaim 只认自己注册的 handler 的 stream（无 handler 任务不启动），mq-starter 不走那套。mq-starter 有自己的 reclaim，配上表的 `mq.redis-stream.reclaim-min-idle` | RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。 diff --git 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 index fa5aa7e..7db3f15 100644 --- 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 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(); } } diff --git 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 new file mode 100644 index 0000000..6d2d638 --- /dev/null +++ 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 MAXLEN ~ }，近似裁剪）。 + * + *

底层中间件的 StreamTrimScheduler 只裁自己注册的 handler 的 stream，而 mq-starter 的消费 + * 不走 handler 体系——driver 不自己裁的话，只经 mq-starter 收发的 topic 无界增长。配置复用中间件的 + * {@code redis-stream.trim.*}（默认开启：maxlen 10 万 / 周期 60s）。 + * + *

两条语义约束：{@code maxlen<=0} 视为关闭——XTRIM 0 会清空整条 stream，绝不能发出； + * 只裁「订阅过」的 stream——与中间件“handler 即消费者”的语义对齐，纯发送方不裁 + * （裁剪节奏该由消费方掌握，发送方不知道还有谁没消费到哪）。 + */ +@Slf4j +class RedisStreamTrimScheduler { + private final StringRedisTemplate redis; + private final RedisStreamProperties props; + private final Set streams = ConcurrentHashMap.newKeySet(); + private ScheduledExecutorService scheduler; + + RedisStreamTrimScheduler(StringRedisTemplate redis, RedisStreamProperties props) { + this.redis = redis; + this.props = props; + } + + /** 订阅时登记 stream；首个登记且 maxlen>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) 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; + } + } +} diff --git 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 new file mode 100644 index 0000000..fb4fff9 --- /dev/null +++ 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 ~）。 + * + *

底层中间件的 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> 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 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> 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> calls = xtrims(); + assertThat(calls).hasSize(2); + assertThat(calls).allSatisfy(c -> { + assertThat(c.subList(2, 5)).as("XTRIM 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> 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 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> 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> awaitXtrim(long timeoutMs) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + List> calls = xtrims(); + if (!calls.isEmpty()) return calls; + Thread.sleep(20); + } + return xtrims(); + } +}