diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java index 1d03965..6a756ac 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java +++ b/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); - } + if (eh != null) { + db.errorHandler(eh); } MqConsumeDispatcher dispatcher = db.build(); + + java.util.function.Consumer 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<payload>[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */ private void consume(MqListenerEndpoint ep, List batch) throws Exception { if (ep.isBatch()) { diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java index b44f974..76a32c6 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java +++ b/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 retryExhaustedHandler; } diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java index 4bb0053..134b043 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java +++ b/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 errors = new ArrayList<>(); + MqConsumeErrorHandler eh = (m, id, ex) -> errors.add(id); + Function 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"); + } } 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 b4f09c6..e6de53d 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 @@ -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 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> 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 claimed; + try { + // StreamOperations(2.3) 无 claim，走 connection 层 xClaim：转移所有权并使 delivery 计数 +1。 + claimed = redis.execute((RedisCallback>) 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 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()); + } } } diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java new file mode 100644 index 0000000..17ec62f --- /dev/null +++ b/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 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; } + } +}