feat(mq-starter): redis-stream 重试可放量（PEL reclaim + 超限落库）

- MqSubscription 增加 retryExhaustedHandler；registry 从 errorHandler 接线 RETRY_EXHAUSTED 回调
- redis driver 每轮 reclaim PEL：connection 层 xClaim 认领超 idle 消息并使 delivery+1
  - delivery <= maxRetry：重投消费，成功 ACK / 失败留 PEL
  - delivery > maxRetry：回调 retryExhaustedHandler 落库 + ACK 终止，不再卡队列
- 可配 reclaimMinIdleMs（默认 5000ms，对齐 legacy 5s 延迟）；4 参构造 + 3 参重载
- 集成测试连真实 redis 验证 重试耗尽→落库→ACK；全仓 28 测试绿

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java

@@ -3,6 +3,7 @@ import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqContext;
 import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqConsumeErrorHandler;
+import com.njcn.mq.spi.MqErrorIdentity;
 import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqIdempotentStore;
 import com.njcn.mq.spi.MqMessageListener;
@@ -33,29 +34,39 @@ public class MqListenerRegistry implements SmartLifecycle {
         if (running) return;
         for (MqListenerEndpoint ep : endpoints) {
             MqListener meta = ep.getMeta();
+            MqConsumeErrorHandler eh = resolveErrorHandler(meta);
+
             MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
                     .batchSize(meta.batchSize())
                     .consumer(batch -> consume(ep, batch));
             if (meta.idempotent()) {
                 db.idempotent(true).store(store);
             }
-            if (errorHandlerResolver != null && !meta.errorHandler().isEmpty()) {
-                MqConsumeErrorHandler eh = errorHandlerResolver.apply(meta.errorHandler());
             if (eh != null) {
                 db.errorHandler(eh);
             }
-            }
             MqConsumeDispatcher dispatcher = db.build();
+
+            java.util.function.Consumer<MqMessage> retryExhausted = eh == null ? null
+                    : msg -> eh.onError(msg, MqErrorIdentity.RETRY_EXHAUSTED, null);
+
             MqSubscription sub = MqSubscription.builder()
                     .topic(meta.topic()).group(meta.group()).tag(meta.tag())
                     .payloadType(ep.getPayloadType())
                     .concurrency(meta.concurrency()).batchSize(meta.batchSize()).maxRetry(meta.maxRetry())
-                    .listener(dispatcher::dispatch).build();
+                    .listener(dispatcher::dispatch)
+                    .retryExhaustedHandler(retryExhausted)
+                    .build();
             driver.subscribe(sub);
         }
         running = true;
     }
 
+    private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) {
+        if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null;
+        return errorHandlerResolver.apply(meta.errorHandler());
+    }
+
     /** 业务调用适配：批量 endpoint 调 (List&lt;payload&gt;[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */
     private void consume(MqListenerEndpoint ep, List<MqMessage> batch) throws Exception {
         if (ep.isBatch()) {

mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java

@@ -1,6 +1,8 @@
 package com.njcn.mq.spi;
+import com.njcn.mq.core.MqMessage;
 import lombok.Builder;
 import lombok.Data;
+import java.util.function.Consumer;
 
 @Data
 @Builder
@@ -13,4 +15,6 @@ public class MqSubscription {
     private int batchSize;
     private int maxRetry;
     private MqMessageListener listener;
+    /** 重试耗尽（超过 maxRetry）时由 driver 回调，用于落库 RETRY_EXHAUSTED；可空。 */
+    private Consumer<MqMessage> retryExhaustedHandler;
 }

mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java

@@ -61,6 +61,13 @@ class MqListenerRegistryTest {
             catch (Exception e) { throw new RuntimeException(e); }
         }
     }
+    /** fake driver：捕获 subscription，不喂消息。 */
+    static class CaptureDriver implements MqDriver {
+        MqSubscription captured;
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() {}
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
     /** fake driver：subscribe 时按序喂多条消息。 */
     static class FeedDriver implements MqDriver {
         private final MqMessage[] msgs;
@@ -123,4 +130,23 @@ class MqListenerRegistryTest {
         assertEquals(1, errors.size(), "按名解析的 errorHandler 应被调用");
         assertEquals(MqErrorIdentity.SINGLE, errors.get(0));
     }
+
+    @Test
+    void retryExhaustedHandler_wiredFromErrorHandler() {
+        ErrBiz biz = new ErrBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        MqConsumeErrorHandler eh = (m, id, ex) -> errors.add(id);
+        Function<String, MqConsumeErrorHandler> resolver = name -> "myEH".equals(name) ? eh : null;
+        CaptureDriver driver = new CaptureDriver();
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints(), null, resolver);
+
+        reg.start();
+
+        assertNotNull(driver.captured.getRetryExhaustedHandler(), "应从 errorHandler 接线重试耗尽回调");
+        driver.captured.getRetryExhaustedHandler().accept(MqMessage.of("k", "", "p"));
+        assertEquals(1, errors.size());
+        assertEquals(MqErrorIdentity.RETRY_EXHAUSTED, errors.get(0), "重试耗尽应落库 RETRY_EXHAUSTED");
+    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -9,6 +9,9 @@ import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqSubscription;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.stream.PendingMessage;
+import org.springframework.data.redis.connection.stream.PendingMessages;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import java.nio.charset.StandardCharsets;
@@ -20,11 +23,17 @@ public class RedisStreamMqDriver implements MqDriver {
     private final StringRedisTemplate redis;
     private final StreamKeyBuilder keyBuilder;
     private final RedisStreamProperties props;
+    private final long reclaimMinIdleMs;
     private final List<Thread> workers = new ArrayList<>();
     private volatile boolean running = true;
 
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
+        this(redis, keyBuilder, props, 5000L);
+    }
+
+    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
         this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
+        this.reclaimMinIdleMs = reclaimMinIdleMs;
     }
 
     @Override public void send(String topic, MqMessage message) {
@@ -54,6 +63,7 @@ public class RedisStreamMqDriver implements MqDriver {
         int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : props.getMaxRetry();
         while (running && !Thread.currentThread().isInterrupted()) {
             try {
+                reclaimPending(sub, stream, group, consumer, maxRetry);
                 List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
                         Consumer.from(group, consumer),
                         StreamReadOptions.empty().count(props.getReadCount()).block(Duration.ofMillis(props.getBlockMs())),
@@ -88,7 +98,52 @@ public class RedisStreamMqDriver implements MqDriver {
             return true;
         } catch (Exception e) {
             log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e);
-            return false; // 留 PEL；超 maxRetry 的兜底由后续 autoclaim 增强补（轻量版先靠重投）
+            return false; // 留 PEL，由 reclaimPending 重投或落库
+        }
+    }
+
+    /**
+     * 处理 PEL 中超过 {@code reclaimMinIdleMs} 仍未 ACK 的消息：
+     * delivery 次数未超 maxRetry 则重投消费；超过则回调 retryExhaustedHandler 落库并 ACK 终止。
+     */
+    private void reclaimPending(MqSubscription sub, String stream, String group, String consumer, int maxRetry) {
+        PendingMessages pendings;
+        try {
+            pendings = redis.opsForStream().pending(stream, group, Range.unbounded(), props.getReadCount());
+        } catch (Exception e) {
+            return; // group 尚未建立或暂无 PEL，忽略
+        }
+        if (pendings == null || pendings.isEmpty()) return;
+        Duration minIdle = Duration.ofMillis(reclaimMinIdleMs);
+        for (PendingMessage pm : pendings) {
+            if (pm.getElapsedTimeSinceLastDelivery().toMillis() < reclaimMinIdleMs) continue;
+            long deliveryCount = pm.getTotalDeliveryCount();
+            List<ByteRecord> claimed;
+            try {
+                // StreamOperations(2.3) 无 claim，走 connection 层 xClaim：转移所有权并使 delivery 计数 +1。
+                claimed = redis.execute((RedisCallback<List<ByteRecord>>) conn ->
+                        conn.streamCommands().xClaim(stream.getBytes(StandardCharsets.UTF_8), group, consumer, minIdle, pm.getId()));
+            } catch (Exception e) { continue; }
+            if (claimed == null || claimed.isEmpty()) continue; // 被其它消费者抢先 claim 或消息已删
+            ByteRecord rec = claimed.get(0);
+            Map<String, String> fields = new HashMap<>();
+            rec.getValue().forEach((k, v) -> fields.put(
+                    new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8)));
+            if (deliveryCount > maxRetry) {
+                if (sub.getRetryExhaustedHandler() != null) {
+                    try {
+                        sub.getRetryExhaustedHandler().accept(MqMessage.of(
+                                fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), null));
+                    } catch (Exception ex) {
+                        log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
+                    }
+                }
+                redis.opsForStream().acknowledge(stream, group, rec.getId());
+                log.warn("[mq-redis] retry exhausted, acked key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount);
+            } else {
+                boolean ack = dispatch(sub, fields, maxRetry);
+                if (ack) redis.opsForStream().acknowledge(stream, group, rec.getId());
+            }
         }
     }
 

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java

@@ -0,0 +1,65 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqSubscription;
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisStreamMqDriverRetryIT {
+
+    @Test
+    void retryExhausted_afterMaxRetry_callsHandlerAndAcks() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-retry");
+        props.setReadCount(10);
+        props.setBlockMs(300);
+        props.setMaxRetry(2);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
+
+        String topic = "MQ_RETRY_IT_" + System.nanoTime();
+        AtomicInteger attempts = new AtomicInteger();
+        BlockingQueue<String> exhausted = new LinkedBlockingQueue<>();
+        driver.subscribe(MqSubscription.builder()
+                .topic(topic).group("g1").tag("*").payloadType(String.class)
+                .concurrency(1).batchSize(1).maxRetry(2)
+                .listener(m -> { attempts.incrementAndGet(); throw new RuntimeException("always fail"); })
+                .retryExhaustedHandler(m -> exhausted.add(m.getKey()))
+                .build());
+
+        driver.send(topic, MqMessage.of("rk1", "", "data"));
+
+        String key = exhausted.poll(12, TimeUnit.SECONDS);
+        assertEquals("rk1", key, "重试耗尽应回调 retryExhaustedHandler");
+        assertTrue(attempts.get() >= 2, "应实际重试若干次后才耗尽, attempts=" + attempts.get());
+
+        Thread.sleep(500);
+        PendingMessagesSummary summary = redis.opsForStream().pending(kb.buildStream(topic), "g1");
+        assertEquals(0L, summary.getTotalPendingMessages(), "耗尽后应 ACK，PEL 清空");
+        driver.stop();
+    }
+
+    private static StringRedisTemplate newRedisOrSkip() {
+        try {
+            LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
+            cf.afterPropertiesSet();
+            StringRedisTemplate t = new StringRedisTemplate(cf);
+            t.afterPropertiesSet();
+            t.hasKey("ping");
+            return t;
+        } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
+    }
+}