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 index 6f2622c..c6a81f4 100644 --- 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 @@ -25,6 +25,8 @@ public class MqConsumeDispatcher { private final MqConsumeErrorHandler errorHandler; private final MqBatchConsumer consumer; private final List buffer = new ArrayList<>(); + /** 串行化三条刷出路径(满批/超时/停机)的业务调用段:业务批量方法任一时刻至多一个调用在跑。 */ + private final Object flushLock = new Object(); private MqConsumeDispatcher(Builder b) { this.batchSize = b.batchSize; @@ -73,11 +75,21 @@ public class MqConsumeDispatcher { } } if (flush != null) { + doFlush(flush); + } + } + + /** + * 刷出一批:业务消费 → 成功逐条 markSuccess / 失败逐条 SINGLE 落库 + 吞。 + * 满批 / 超时 / 停机三条路径都汇聚到这里,失败语义只写一份: + * 批量消息已逐条 ACK 无法精确重投整批 → 对齐 legacy 不卡队列:落库 + 吞掉, + * 不抛、不标 success(去重 processing 标记自然过期)。 + */ + private void doFlush(List flush) { + synchronized (flushLock) { 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);