Middleware/mq-starter
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mq-starter): 落实 opus review 反馈（含 1 个 Critical 静默丢消息）

Browse Source 
- [Critical] dispatcher errorHandler 抛异常时仍保证 markFail + 抛原始异常（safeOnError 隔离）：
  否则去重标记停在 processing，reclaim 重投被去重挡掉并 ACK → 静默丢消息
- [Important] 重试耗尽回调携带完整 payload（reclaim 解码 body，不再传 null）
- [Important] reclaim 默认 idle 5s→30s，避免多实例误 reclaim in-flight 消息；注释 at-least-once 语义
- [Important] idempotent=true 但无 MqIdempotentStore bean 时 start() 告警
- core 引入 slf4j-api(provided)；批量失败 / errorHandler 失败均记日志
- errorHandler bean 未找到时抛带 topic/group 的清晰错误
- 文档化已知限制（批量无 idle flush、去重 fail→reopen 非原子）
- 全仓 29 测试绿（含连真实 redis 的 IT）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
hongawen
2026-06-29 20:48:53 +08:00
parent dd2b933c08
commit f0dc785998
7 changed files with 81 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
mq-starter-core/pom.xml
View File

@@ -8,6 +8,7 @@
<dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
<dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.30</version><scope>provided</scope></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><version>${springboot.version}</version><optional>true</optional></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
</dependencies>

28
mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java
View File

@@ -4,6 +4,7 @@ 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;
@@ -13,8 +14,10 @@ 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;
@@ -38,13 +41,13 @@ public class MqConsumeDispatcher {
try {
consumer.consume(Collections.singletonList(msg));
} catch (Exception e) {
if (errorHandler != null) {
errorHandler.onError(msg, MqErrorIdentity.SINGLE, e);
}
// errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing，
// reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。
safeOnError(msg, MqErrorIdentity.SINGLE, e);
if (idempotent && store != null) {
store.markFail(msg.getKey());
}
throw e; // 交 driver 重投
throw e; // 抛原始异常，交 driver 重投
}
if (idempotent && store != null) {
store.markSuccess(msg.getKey());
@@ -66,10 +69,9 @@ public class MqConsumeDispatcher {
} catch (Exception e) {
// 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列：
// 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。
if (errorHandler != null) {
for (MqMessage m : flush) {
errorHandler.onError(m, MqErrorIdentity.SINGLE, e);
}
log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e);
for (MqMessage m : flush) {
safeOnError(m, MqErrorIdentity.SINGLE, e);
}
return;
}
@@ -81,6 +83,16 @@ public class MqConsumeDispatcher {
}
}
/** 调用异常落库回调，并隔离其自身异常（落库是 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 {

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

@@ -8,10 +8,12 @@ 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<MqListenerEndpoint> endpoints;
@@ -34,6 +36,10 @@ public class MqListenerRegistry implements SmartLifecycle {
if (running) return;
for (MqListenerEndpoint ep : endpoints) {
MqListener meta = ep.getMeta();
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()
@@ -64,7 +70,12 @@ public class MqListenerRegistry implements SmartLifecycle {
private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) {
if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null;
return errorHandlerResolver.apply(meta.errorHandler());
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&lt;payload&gt;[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */

24
mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java
View File

@@ -142,4 +142,28 @@ class MqConsumeDispatcherTest {
assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success");
}
@Test
void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() {
List<String> 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 时被去重挡掉而丢消息");
}
}

2
mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java
View File

@@ -41,6 +41,8 @@ public class RedisMqIdempotentStore implements MqIdempotentStore {
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;

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

@@ -28,7 +28,9 @@ public class RedisStreamMqDriver implements MqDriver {
private volatile boolean running = true;
public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
this(redis, keyBuilder, props, 5000L);
// 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) {
@@ -132,8 +134,7 @@ public class RedisStreamMqDriver implements MqDriver {
if (deliveryCount > maxRetry) {
if (sub.getRetryExhaustedHandler() != null) {
try {
sub.getRetryExhaustedHandler().accept(MqMessage.of(
fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), null));
sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields));
} catch (Exception ex) {
log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
}
@@ -147,6 +148,18 @@ public class RedisStreamMqDriver implements MqDriver {
}
}
/** 解码消息体为业务 payload（best-effort：毒消息时 payload 为 null），用于重试耗尽落库回调。 */
private MqMessage buildMessage(MqSubscription sub, Map<String, String> 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<Object>) c -> c.execute("XGROUP",

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

@@ -32,18 +32,20 @@ class RedisStreamMqDriverRetryIT {
String topic = "MQ_RETRY_IT_" + System.nanoTime();
AtomicInteger attempts = new AtomicInteger();
BlockingQueue<String> exhausted = new LinkedBlockingQueue<>();
BlockingQueue<MqMessage> 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()))
.retryExhaustedHandler(exhausted::add)
.build());
driver.send(topic, MqMessage.of("rk1", "", "data"));
String key = exhausted.poll(12, TimeUnit.SECONDS);
assertEquals("rk1", key, "重试耗尽应回调 retryExhaustedHandler");
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);