diff --git a/mq-starter-core/pom.xml b/mq-starter-core/pom.xml index 81c26d0..a63dd49 100644 --- a/mq-starter-core/pom.xml +++ b/mq-starter-core/pom.xml @@ -8,6 +8,7 @@ com.alibabafastjson${fastjson.version} cn.hutoolhutool-all${hutool.version} org.projectlomboklombok${lombok.version}true + org.slf4jslf4j-api1.7.30provided org.springframework.bootspring-boot-configuration-processor${springboot.version}true org.springframework.bootspring-boot-starter-test${springboot.version}test diff --git a/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java index f81ebeb..303ff03 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java @@ -13,4 +13,8 @@ public @interface MqListener { int batchSize() default 1; int maxRetry() default 3; String tag() default "*"; + /** 是否启用消费去重（按 key）。默认关闭。 */ + boolean idempotent() default false; + /** 异常落库回调的 bean 名；空表示不落库。 */ + String errorHandler() default ""; } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqCoreAutoConfiguration.java b/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqCoreAutoConfiguration.java index ca454d7..0bbc32c 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqCoreAutoConfiguration.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqCoreAutoConfiguration.java @@ -3,7 +3,9 @@ import com.njcn.mq.container.MqListenerAnnotationBeanPostProcessor; import com.njcn.mq.container.MqListenerRegistry; import com.njcn.mq.core.DefaultMqTemplate; import com.njcn.mq.core.MqTemplate; +import com.njcn.mq.spi.MqConsumeErrorHandler; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqIdempotentStore; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.context.annotation.Bean; @@ -27,7 +29,13 @@ public class MqCoreAutoConfiguration { @Bean @ConditionalOnBean(MqDriver.class) @ConditionalOnMissingBean - public MqListenerRegistry mqListenerRegistry(MqDriver driver, MqListenerAnnotationBeanPostProcessor bpp) { - return new MqListenerRegistry(driver, bpp.getEndpoints()); + public MqListenerRegistry mqListenerRegistry(MqDriver driver, + MqListenerAnnotationBeanPostProcessor bpp, + org.springframework.beans.factory.ObjectProvider storeProvider, + org.springframework.context.ApplicationContext appCtx) { + MqIdempotentStore store = storeProvider.getIfAvailable(); + java.util.function.Function resolver = + name -> appCtx.getBean(name, MqConsumeErrorHandler.class); + return new MqListenerRegistry(driver, bpp.getEndpoints(), store, resolver); } } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqBatchConsumer.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqBatchConsumer.java new file mode 100644 index 0000000..5c647cc --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqBatchConsumer.java @@ -0,0 +1,14 @@ +package com.njcn.mq.container; + +import com.njcn.mq.core.MqMessage; +import java.util.List; + +/** + * 业务消费回调（core 内部契约）。 + * 由 {@link MqListenerRegistry} 把 @MqListener 方法的反射调用适配成统一的「批量入参」形态： + * 单条流传 size=1 的 list，批量流传攒满的 list。dispatcher 只面向它编排，不关心反射细节。 + */ +@FunctionalInterface +public interface MqBatchConsumer { + void consume(List batch) throws Exception; +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java new file mode 100644 index 0000000..8ede99d --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java @@ -0,0 +1,111 @@ +package com.njcn.mq.container; + +import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqConsumeErrorHandler; +import com.njcn.mq.spi.MqErrorIdentity; +import com.njcn.mq.spi.MqIdempotentStore; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。 + * 纯逻辑、不依赖 spring / redis，便于单测。重试本身交给 driver 的 MQ 原生重投。 + */ +@Slf4j +public class MqConsumeDispatcher { + + // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。 + private final int batchSize; + private final boolean idempotent; + private final MqIdempotentStore store; + private final MqConsumeErrorHandler errorHandler; + private final MqBatchConsumer consumer; + private final List buffer = new ArrayList<>(); + + private MqConsumeDispatcher(Builder b) { + this.batchSize = b.batchSize; + this.idempotent = b.idempotent; + this.store = b.store; + this.errorHandler = b.errorHandler; + this.consumer = b.consumer; + } + + public void dispatch(MqMessage msg) throws Exception { + if (idempotent && store != null && !store.tryAcquire(msg.getKey())) { + return; // processing 中或已 success，跳过 + } + if (batchSize <= 1) { + try { + consumer.consume(Collections.singletonList(msg)); + } catch (Exception e) { + // errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing， + // reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。 + safeOnError(msg, MqErrorIdentity.SINGLE, e); + if (idempotent && store != null) { + store.markFail(msg.getKey()); + } + throw e; // 抛原始异常，交 driver 重投 + } + if (idempotent && store != null) { + store.markSuccess(msg.getKey()); + } + return; + } + // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。 + List flush = null; + synchronized (buffer) { + buffer.add(msg); + if (buffer.size() >= batchSize) { + flush = new ArrayList<>(buffer); + buffer.clear(); + } + } + if (flush != null) { + try { + consumer.consume(flush); + } catch (Exception e) { + // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列： + // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。 + log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e); + for (MqMessage m : flush) { + safeOnError(m, MqErrorIdentity.SINGLE, e); + } + return; + } + if (idempotent && store != null) { + for (MqMessage m : flush) { + store.markSuccess(m.getKey()); + } + } + } + } + + /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */ + private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) { + if (errorHandler == null) return; + try { + errorHandler.onError(msg, identity, cause); + } catch (Exception ehEx) { + log.error("[mq] errorHandler 落库失败 key={}（不影响主流程）", msg.getKey(), ehEx); + } + } + + public static Builder builder() { return new Builder(); } + + public static class Builder { + private int batchSize = 1; + private boolean idempotent = false; + private MqIdempotentStore store; + private MqConsumeErrorHandler errorHandler; + private MqBatchConsumer consumer; + public Builder batchSize(int n) { this.batchSize = n; return this; } + public Builder idempotent(boolean v) { this.idempotent = v; return this; } + public Builder store(MqIdempotentStore s) { this.store = s; return this; } + public Builder errorHandler(MqConsumeErrorHandler h) { this.errorHandler = h; return this; } + public Builder consumer(MqBatchConsumer c) { this.consumer = c; return this; } + public MqConsumeDispatcher build() { return new MqConsumeDispatcher(this); } + } +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java index 72d3125..a125b80 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java @@ -26,9 +26,21 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor if (wantsContext && params[1] != MqContext.class) { throw new IllegalStateException("@MqListener 第二参须为 MqContext: " + m); } - Class payloadType = params[0]; + Class first = params[0]; + boolean batch = List.class.isAssignableFrom(first); + Class payloadType; + if (batch) { + // 批量方法 (List)：取泛型实参 T 作为 payloadType，driver 仍按 T 逐条反序列化，dispatcher 负责攒批。 + java.lang.reflect.Type gt = m.getGenericParameterTypes()[0]; + if (!(gt instanceof java.lang.reflect.ParameterizedType)) { + throw new IllegalStateException("@MqListener 批量方法须声明具体元素类型 List: " + m); + } + payloadType = (Class) ((java.lang.reflect.ParameterizedType) gt).getActualTypeArguments()[0]; + } else { + payloadType = first; + } m.setAccessible(true); - endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext)); + endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch)); } return bean; } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java index 366b562..958f1f6 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java @@ -12,4 +12,6 @@ public class MqListenerEndpoint { private MqListener meta; private Class payloadType; private boolean wantsContext; + /** true=方法签名为 (List<T>)，批量消费；payloadType 此时为元素类型 T。 */ + private boolean batch; } 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 b912f9d..2e74217 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 @@ -2,40 +2,104 @@ package com.njcn.mq.container; 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; import com.njcn.mq.spi.MqSubscription; +import lombok.extern.slf4j.Slf4j; import org.springframework.context.SmartLifecycle; +import java.util.ArrayList; import java.util.List; +@Slf4j public class MqListenerRegistry implements SmartLifecycle { private final MqDriver driver; private final List endpoints; + private final MqIdempotentStore store; + private final java.util.function.Function errorHandlerResolver; private volatile boolean running = false; public MqListenerRegistry(MqDriver driver, List endpoints) { + this(driver, endpoints, null, null); + } + + public MqListenerRegistry(MqDriver driver, List endpoints, + MqIdempotentStore store, + java.util.function.Function errorHandlerResolver) { this.driver = driver; this.endpoints = endpoints; + this.store = store; this.errorHandlerResolver = errorHandlerResolver; } @Override public synchronized void start() { if (running) return; for (MqListenerEndpoint ep : endpoints) { MqListener meta = ep.getMeta(); - MqMessageListener listener = (MqMessage msg) -> invoke(ep, msg); + if (meta.idempotent() && store == null) { + log.warn("[mq] @MqListener(idempotent=true) 但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}", + meta.topic(), meta.group()); + } + 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 (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(listener).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; + try { + return errorHandlerResolver.apply(meta.errorHandler()); + } catch (Exception e) { + throw new IllegalStateException("@MqListener(errorHandler=\"" + meta.errorHandler() + + "\") 指定的 bean 未找到: topic=" + meta.topic() + " group=" + meta.group(), e); + } + } + + /** 业务调用适配：批量 endpoint 调 (List<payload>[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */ + private void consume(MqListenerEndpoint ep, List batch) throws Exception { + if (ep.isBatch()) { + List payloads = new ArrayList<>(batch.size()); + for (MqMessage m : batch) payloads.add(m.getPayload()); + Object[] args = ep.isWantsContext() + ? new Object[]{ payloads, new MqContext(batch.get(batch.size() - 1)) } + : new Object[]{ payloads }; + invokeReflect(ep, args); + } else { + for (MqMessage m : batch) invoke(ep, m); + } + } + private void invoke(MqListenerEndpoint ep, MqMessage msg) throws Exception { Object[] args = ep.isWantsContext() ? new Object[]{ msg.getPayload(), new MqContext(msg) } : new Object[]{ msg.getPayload() }; + invokeReflect(ep, args); + } + + private void invokeReflect(MqListenerEndpoint ep, Object[] args) throws Exception { try { ep.getMethod().invoke(ep.getBean(), args); } catch (java.lang.reflect.InvocationTargetException e) { Throwable c = e.getCause(); diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqConsumeErrorHandler.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqConsumeErrorHandler.java new file mode 100644 index 0000000..c602dad --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqConsumeErrorHandler.java @@ -0,0 +1,18 @@ +package com.njcn.mq.spi; + +import com.njcn.mq.core.MqMessage; + +/** + * 消费异常落库回调 SPI。 + * + *

业务实现（如调 feign 写错误日志表）；core 默认 no-op。对齐 legacy 的 saveExceptionMsgLog。 + */ +public interface MqConsumeErrorHandler { + + /** + * @param message 失败的消息 + * @param identity {@link MqErrorIdentity#SINGLE} 单次失败 / {@link MqErrorIdentity#RETRY_EXHAUSTED} 重试耗尽 + * @param exception 失败异常（RETRY_EXHAUSTED 由 driver 触发时可能为 null） + */ + void onError(MqMessage message, MqErrorIdentity identity, Exception exception); +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqErrorIdentity.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqErrorIdentity.java new file mode 100644 index 0000000..e400fd1 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqErrorIdentity.java @@ -0,0 +1,9 @@ +package com.njcn.mq.spi; + +/** 消费异常落库的身份标识，对齐 legacy 的 IDENTITY_SINGLE / IDENTITY_RETRY。 */ +public enum MqErrorIdentity { + /** 单次消费失败。 */ + SINGLE, + /** 重试耗尽（超过 maxRetry）。 */ + RETRY_EXHAUSTED +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java new file mode 100644 index 0000000..b04c4f8 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java @@ -0,0 +1,24 @@ +package com.njcn.mq.spi; + +/** + * 消费去重 SPI（按消息 key 维度）。 + * + *

状态机:processing / success / fail。语义对齐 legacy 的 filter+consumeSuccess+saveException, + * 但修正了 legacy 读写 key 不一致的缺陷(默认实现读写同一前缀键)。 + */ +public interface MqIdempotentStore { + + /** + * 尝试占用消费资格。 + * + * @return true=可消费(并已标记 processing);false=应跳过(processing 中或已 success)。 + * 无记录或上次 fail 时放行。 + */ + boolean tryAcquire(String key); + + /** 消费成功,标记 success(较长 TTL,防重复消费)。 */ + void markSuccess(String key); + + /** 消费失败,标记 fail(允许后续重新消费)。 */ + void markFail(String key); +} 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/autoconfig/MqCoreAutoConfigurationTest.java b/mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqCoreAutoConfigurationTest.java index 3b2e210..6b3c3c1 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqCoreAutoConfigurationTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqCoreAutoConfigurationTest.java @@ -1,14 +1,19 @@ package com.njcn.mq.autoconfig; +import com.njcn.mq.annotation.MqListener; import com.njcn.mq.core.MqMessage; import com.njcn.mq.core.MqTemplate; import com.njcn.mq.container.MqListenerRegistry; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqIdempotentStore; import com.njcn.mq.spi.MqSubscription; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; class MqCoreAutoConfigurationTest { @@ -32,4 +37,44 @@ class MqCoreAutoConfigurationTest { .withConfiguration(AutoConfigurations.of(MqCoreAutoConfiguration.class)) .run(ctx -> assertThat(ctx).doesNotHaveBean(MqTemplate.class)); } + + static class InMemoryStore implements MqIdempotentStore { + final Map m = new ConcurrentHashMap<>(); + public boolean tryAcquire(String key) { + String s = m.get(key); + if (s == null || "fail".equals(s)) { m.put(key, "processing"); return true; } + return false; + } + public void markSuccess(String key) { m.put(key, "success"); } + public void markFail(String key) { m.put(key, "fail"); } + } + static class DedupListener { + final AtomicInteger count = new AtomicInteger(); + @MqListener(topic = "D", group = "dg", idempotent = true) + public void on(String msg) { count.incrementAndGet(); } + } + @Configuration static class IdempotentSetup { + @Bean MqDriver driver() { return new MqDriver() { + public void send(String t, MqMessage m) {} + public void subscribe(MqSubscription s) { + try { + s.getListener().onMessage(MqMessage.of("dup", "", "x")); + s.getListener().onMessage(MqMessage.of("dup", "", "y")); + } catch (Exception e) { throw new RuntimeException(e); } + } + public void stop() {} }; } + @Bean MqIdempotentStore store() { return new InMemoryStore(); } + @Bean DedupListener dedupListener() { return new DedupListener(); } + } + @Test + void idempotentDedup_endToEnd_whenStoreBeanPresent() { + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(MqCoreAutoConfiguration.class)) + .withUserConfiguration(IdempotentSetup.class) + .run(ctx -> { + assertThat(ctx).hasNotFailed(); + DedupListener l = ctx.getBean(DedupListener.class); + assertThat(l.count.get()).isEqualTo(1); + }); + } } diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java new file mode 100644 index 0000000..1febbfd --- /dev/null +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java @@ -0,0 +1,169 @@ +package com.njcn.mq.container; + +import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqConsumeErrorHandler; +import com.njcn.mq.spi.MqErrorIdentity; +import com.njcn.mq.spi.MqIdempotentStore; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +class MqConsumeDispatcherTest { + + @Test + void singleMode_invokesConsumerWithMessage() throws Exception { + List> consumed = new ArrayList<>(); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1) + .consumer(consumed::add) + .build(); + + d.dispatch(MqMessage.of("k1", "", "hello")); + + assertEquals(1, consumed.size()); + assertEquals("hello", consumed.get(0).get(0).getPayload()); + } + + @Test + void batchMode_flushesOnceWhenBufferReachesBatchSize() throws Exception { + List> consumed = new ArrayList<>(); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(3) + .consumer(consumed::add) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); + d.dispatch(MqMessage.of("k2", "", "b")); + assertEquals(0, consumed.size(), "未满批不应消费"); + + d.dispatch(MqMessage.of("k3", "", "c")); + assertEquals(1, consumed.size(), "满批触发一次消费"); + assertEquals(3, consumed.get(0).size()); + } + + @Test + void idempotent_skipsWhenStoreRejects() throws Exception { + List> consumed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return !"dup".equals(key); } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store) + .consumer(consumed::add) + .build(); + + d.dispatch(MqMessage.of("ok", "", "a")); + d.dispatch(MqMessage.of("dup", "", "b")); + + assertEquals(1, consumed.size(), "被拒绝的 key 应跳过"); + assertEquals("a", consumed.get(0).get(0).getPayload()); + } + + @Test + void idempotent_marksSuccessAfterConsume() throws Exception { + List succeeded = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return true; } + @Override public void markSuccess(String key) { succeeded.add(key); } + @Override public void markFail(String key) { } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store) + .consumer(batch -> { }) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); + + assertEquals(1, succeeded.size(), "消费成功后应标记 success"); + assertEquals("k1", succeeded.get(0)); + } + + @Test + void singleMode_onException_callsErrorHandlerMarksFailAndRethrows() { + List errors = new ArrayList<>(); + List failed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return true; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { failed.add(key); } + }; + MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity); + RuntimeException boom = new RuntimeException("boom"); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store).errorHandler(eh) + .consumer(batch -> { throw boom; }) + .build(); + + Exception thrown = assertThrows(Exception.class, + () -> d.dispatch(MqMessage.of("k1", "", "a"))); + + assertSame(boom, thrown, "原异常应抛出交 driver 重投"); + assertEquals(1, errors.size()); + assertEquals(MqErrorIdentity.SINGLE, errors.get(0), "单次失败应落库 SINGLE"); + assertEquals(1, failed.size()); + assertEquals("k1", failed.get(0), "失败应标记 fail 允许重新消费"); + } + + @Test + void batchMode_onException_swallowsAndCallsErrorHandler() throws Exception { + List errors = new ArrayList<>(); + MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(2).errorHandler(eh) + .consumer(batch -> { throw new RuntimeException("db down"); }) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); // 攒批，未满 + assertDoesNotThrow(() -> d.dispatch(MqMessage.of("k2", "", "b")), + "批量入库失败不应抛出（对齐 legacy 不卡队列）"); + assertEquals(2, errors.size(), "批量失败应对每条落库"); + } + + @Test + void batchMode_marksSuccessForEachAfterFlush() throws Exception { + List succeeded = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return true; } + @Override public void markSuccess(String key) { succeeded.add(key); } + @Override public void markFail(String key) { } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(2).idempotent(true).store(store) + .consumer(batch -> { }) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); + d.dispatch(MqMessage.of("k2", "", "b")); + + assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success"); + } + + @Test + void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() { + List failed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return true; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { failed.add(key); } + }; + // errorHandler 自身抛异常（落库 sink 挂掉，常与错误风暴同时发生） + MqConsumeErrorHandler eh = (m, id, ex) -> { throw new RuntimeException("sink down"); }; + RuntimeException boom = new RuntimeException("biz boom"); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store).errorHandler(eh) + .consumer(batch -> { throw boom; }) + .build(); + + Exception thrown = assertThrows(Exception.class, + () -> d.dispatch(MqMessage.of("k1", "", "a"))); + + assertSame(boom, thrown, "errorHandler 失败不应掩盖原始消费异常"); + assertEquals(1, failed.size(), + "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息"); + } +} 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 166a078..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 @@ -2,9 +2,19 @@ package com.njcn.mq.container; 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.MqDriver; +import com.njcn.mq.spi.MqErrorIdentity; +import com.njcn.mq.spi.MqIdempotentStore; import com.njcn.mq.spi.MqSubscription; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; @@ -15,6 +25,33 @@ class MqListenerRegistryTest { @MqListener(topic = "LN_Topic", group = "ln_consumer") public void on(String payload, MqContext ctx) { got.set(payload); ctxKey.set(ctx.getKey()); } } + static class BatchBiz { + final List received = new ArrayList<>(); + @MqListener(topic = "B", group = "bg", batchSize = 2) + public void onBatch(List msgs) { received.addAll(msgs); } + } + static class DedupBiz { + final AtomicInteger count = new AtomicInteger(); + @MqListener(topic = "D", group = "dg", idempotent = true) + public void on(String msg) { count.incrementAndGet(); } + } + static class ErrBiz { + @MqListener(topic = "E", group = "eg", errorHandler = "myEH") + public void on(String msg) { throw new RuntimeException("biz fail"); } + } + + /** 真实状态机的内存去重 store。 */ + static class InMemoryStore implements MqIdempotentStore { + final Map m = new ConcurrentHashMap<>(); + @Override public boolean tryAcquire(String key) { + String s = m.get(key); + if (s == null || "fail".equals(s)) { m.put(key, "processing"); return true; } + return false; + } + @Override public void markSuccess(String key) { m.put(key, "success"); } + @Override public void markFail(String key) { m.put(key, "fail"); } + } + /** fake driver：subscribe 时立刻把一条消息喂给 listener。 */ static class ImmediateDriver implements MqDriver { @Override public void send(String topic, MqMessage m) {} @@ -24,6 +61,24 @@ 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; + FeedDriver(MqMessage... msgs) { this.msgs = msgs; } + @Override public void send(String topic, MqMessage m) {} + @Override public void stop() {} + @Override public void subscribe(MqSubscription sub) { + try { for (MqMessage m : msgs) sub.getListener().onMessage(m); } + catch (Exception e) { throw new RuntimeException(e); } + } + } @Test void start_subscribesAndInvokesBusinessMethodWithPayloadAndContext() throws Exception { @@ -35,4 +90,63 @@ class MqListenerRegistryTest { assertEquals("hi", biz.got.get()); assertEquals("k9", biz.ctxKey.get()); } + + @Test + void batchListener_accumulatesAndInvokesWithPayloadList() { + BatchBiz biz = new BatchBiz(); + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(biz, "biz"); + MqListenerRegistry reg = new MqListenerRegistry( + new FeedDriver(MqMessage.of("k1", "", "a"), MqMessage.of("k2", "", "b")), + bpp.getEndpoints()); + reg.start(); + assertEquals(Arrays.asList("a", "b"), biz.received, "批量方法应收到攒满的 payload 列表"); + } + + @Test + void idempotentListener_dedupsSameKey() { + DedupBiz biz = new DedupBiz(); + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(biz, "biz"); + MqListenerRegistry reg = new MqListenerRegistry( + new FeedDriver(MqMessage.of("dupKey", "", "x"), MqMessage.of("dupKey", "", "y")), + bpp.getEndpoints(), new InMemoryStore(), null); + reg.start(); + assertEquals(1, biz.count.get(), "同 key 重复消息应只消费一次"); + } + + @Test + void errorHandler_resolvedByNameAndCalledOnException() { + 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; + MqListenerRegistry reg = new MqListenerRegistry( + new FeedDriver(MqMessage.of("k1", "", "a")), bpp.getEndpoints(), null, resolver); + + assertThrows(RuntimeException.class, reg::start, "单条异常应向 driver 抛出"); + 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-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java index f65fef2..c1336e2 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java @@ -13,6 +13,11 @@ class MqListenerScanTest { public void onWithCtx(String msg, MqContext ctx) { } } + static class BatchBiz { + @MqListener(topic = "B", group = "bg", batchSize = 10) + public void onBatch(List msgs) { } + } + @Test void scan_findsEndpoints_withPayloadTypeAndContextFlag() { MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); @@ -25,4 +30,13 @@ class MqListenerScanTest { MqListenerEndpoint e2 = eps.stream().filter(e -> e.getMeta().topic().equals("T2")).findFirst().orElseThrow(AssertionError::new); assertTrue(e2.isWantsContext()); } + + @Test + void scan_batchListener_resolvesElementTypeAndMarksBatch() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new BatchBiz(), "biz"); + MqListenerEndpoint ep = bpp.getEndpoints().get(0); + assertTrue(ep.isBatch(), "List 参数应识别为批量"); + assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T"); + } } diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java new file mode 100644 index 0000000..6eecf68 --- /dev/null +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java @@ -0,0 +1,62 @@ +package com.njcn.mq.driver.redis; + +import com.njcn.mq.spi.MqIdempotentStore; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Duration; + +/** + * 基于 Redis 的去重默认实现。状态机 processing / success / fail，各带 TTL。 + * + *

对齐 legacy filter+consumeSuccess+saveException 的语义，但修正了 legacy「读裸 key、写带前缀 key」 + * 导致去重失效的缺陷：本实现读写统一使用 {@code prefix + key}。 + */ +public class RedisMqIdempotentStore implements MqIdempotentStore { + + private static final String PROCESSING = "processing"; + private static final String SUCCESS = "success"; + private static final String FAIL = "fail"; + + private final StringRedisTemplate redis; + private final String prefix; + private final long processingTtl; + private final long successTtl; + private final long failTtl; + + public RedisMqIdempotentStore(StringRedisTemplate redis, String prefix, + long processingTtlSeconds, long successTtlSeconds, long failTtlSeconds) { + this.redis = redis; + this.prefix = prefix; + this.processingTtl = processingTtlSeconds; + this.successTtl = successTtlSeconds; + this.failTtl = failTtlSeconds; + } + + @Override + public boolean tryAcquire(String key) { + String k = prefix + key; + // setIfAbsent 原子占用：key 不存在时写 processing 并放行。 + Boolean acquired = redis.opsForValue().setIfAbsent(k, PROCESSING, Duration.ofSeconds(processingTtl)); + if (Boolean.TRUE.equals(acquired)) { + return true; + } + // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。 + // 注意：fail→reopen 的 get+set 非原子，跨实例对「共享同一 key 的不同消息」可能都放行； + // 同一条消息由 stream PEL/claim 序列化，故仅影响不同消息共享 key 的边缘场景。 + if (FAIL.equals(redis.opsForValue().get(k))) { + redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl)); + return true; + } + return false; + } + + @Override + public void markSuccess(String key) { + redis.opsForValue().set(prefix + key, SUCCESS, Duration.ofSeconds(successTtl)); + } + + @Override + public void markFail(String key) { + redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl)); + } +} 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..5da965d 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,19 @@ 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) { + // reclaim 默认 idle 30s：须大于正常处理耗时，避免多实例误 reclaim 仍在处理(in-flight)的消息造成重复消费； + // 非幂等消费在多实例下为 at-least-once，可通过 4 参构造按业务处理时延调整。 + this(redis, keyBuilder, props, 30_000L); + } + + 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 +65,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,10 +100,66 @@ 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(buildMessage(sub, fields)); + } 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()); + } + } + } + + /** 解码消息体为业务 payload（best-effort：毒消息时 payload 为 null），用于重试耗尽落库回调。 */ + private MqMessage buildMessage(MqSubscription sub, Map fields) { + Object payload = null; + try { + String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY)); + payload = JSON.parseObject(json, sub.getPayloadType()); + } catch (Exception ignore) { + // 毒消息：payload 留 null，至少保留 key/tag 供落库 + } + return MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload); + } + private void ensureGroup(String stream, String group) { try { redis.execute((RedisCallback) c -> c.execute("XGROUP", diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java index e9736a5..4dac641 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java @@ -3,6 +3,7 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties; import com.njcn.middle.stream.support.StreamKeyBuilder; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqIdempotentStore; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -21,4 +22,11 @@ public class RedisStreamMqDriverAutoConfiguration { public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) { return new RedisStreamMqDriver(redis, streamKeyBuilder, props); } + + @Bean + @ConditionalOnMissingBean + public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis) { + // 默认 TTL：processing 60s / success 300s / fail 300s，对齐 legacy 量级；统一前缀（修正 key 一致性）。 + return new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300); + } } diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java new file mode 100644 index 0000000..5dd81a2 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java @@ -0,0 +1,38 @@ +package com.njcn.mq.driver.redis; + +import org.junit.jupiter.api.Test; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; + +import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +class RedisMqIdempotentStoreIT { + + @Test + void stateMachine_firstAcquire_thenBlocked_failReopens_successBlocks() { + StringRedisTemplate redis = newRedisOrSkip(); + RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:it:", 60, 300, 300); + String key = "k_" + System.nanoTime(); + + assertTrue(store.tryAcquire(key), "首次应放行并占用 processing"); + assertFalse(store.tryAcquire(key), "processing 中应挡回"); + + store.markFail(key); + assertTrue(store.tryAcquire(key), "上次 fail 后应允许重新消费"); + + store.markSuccess(key); + assertFalse(store.tryAcquire(key), "success 后应挡回，避免重复消费"); + } + + 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; } + } +} 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..d35d826 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java @@ -0,0 +1,67 @@ +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(exhausted::add) + .build()); + + driver.send(topic, MqMessage.of("rk1", "", "data")); + + MqMessage exhaustedMsg = exhausted.poll(12, TimeUnit.SECONDS); + assertNotNull(exhaustedMsg, "重试耗尽应回调 retryExhaustedHandler"); + assertEquals("rk1", exhaustedMsg.getKey()); + assertEquals("data", exhaustedMsg.getPayload(), "耗尽回调应携带完整 payload，而非 null"); + 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; } + } +} diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java index 3e2663a..953e0de 100644 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java @@ -1,8 +1,13 @@ package com.njcn.mq.driver.redis; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; +import com.njcn.mq.spi.MqIdempotentStore; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.lang.reflect.Method; import static org.assertj.core.api.Assertions.assertThat; @@ -30,4 +35,16 @@ class RedisStreamMqDriverWiringTest { .as("@AutoConfigureBefore 必须指向 MqCoreAutoConfiguration") .contains(MqCoreAutoConfiguration.class); } + + @Test + void declaresIdempotentStoreBean() throws Exception { + Method m = RedisStreamMqDriverAutoConfiguration.class + .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class); + assertThat(m.getAnnotation(Bean.class)) + .as("redis-stream 模式应提供默认去重实现") + .isNotNull(); + assertThat(MqIdempotentStore.class) + .as("默认去重实现须为 MqIdempotentStore 类型") + .isAssignableFrom(m.getReturnType()); + } }