From f4793b10c051e8b9a110c91fca8f3eed61978971 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Fri, 10 Jul 2026 02:54:09 +0100 Subject: [PATCH] =?UTF-8?q?fix(mq-starter-rocketmq):=20=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E6=8D=95=E8=8E=B7=20MqMessageInProgressException=20=E8=BF=94?= =?UTF-8?q?=E5=9B=9E=20RECONSUME=5FLATER=EF=BC=8C=E6=9D=9C=E7=BB=9D?= =?UTF-8?q?=E5=A4=84=E7=90=86=E4=B8=AD=E6=B6=88=E6=81=AF=E8=A2=AB=E8=80=97?= =?UTF-8?q?=E5=B0=BD=E8=90=BD=E5=BA=93+ACK=20=E9=9D=99=E9=BB=98=E4=B8=A2?= =?UTF-8?q?=E5=BC=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- .../mq/driver/rocketmq/RocketMqDriver.java | 6 ++++ .../rocketmq/RocketMqDriverConsumeTest.java | 36 +++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java index 6d40765..12dce2e 100644 --- a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java +++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java @@ -3,6 +3,7 @@ package com.njcn.mq.driver.rocketmq; import com.alibaba.fastjson.JSON; import com.njcn.mq.core.MqMessage; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqMessageInProgressException; import com.njcn.mq.spi.MqSubscription; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; @@ -63,6 +64,11 @@ public class RocketMqDriver implements MqDriver { String json = new String(ext.getBody(), StandardCharsets.UTF_8); Object payload = JSON.parseObject(json, sub.getPayloadType()); sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload)); + } catch (MqMessageInProgressException inProgress) { + // 去重"处理中"信号:不 ACK,交 broker 重投;绝不进耗尽落库分支(否则并发下静默丢消息)。 + // 语义对齐 RedisStreamMqDriver#dispatch 对该异常的 return false(留 PEL)。 + log.debug("[mq-rocket] in-progress, reconsume later key={}", ext.getKeys()); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; } catch (Exception e) { if (ext.getReconsumeTimes() >= maxRetry) { // 重试耗尽:回调 retryExhaustedHandler 落库并 ACK 终止,不再无限重投 diff --git a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java index 0f0fcad..e3d3cd2 100644 --- a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java +++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java @@ -1,6 +1,7 @@ package com.njcn.mq.driver.rocketmq; import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqMessageInProgressException; import com.njcn.mq.spi.MqSubscription; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.common.message.MessageExt; @@ -84,4 +85,39 @@ class RocketMqDriverConsumeTest { assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status); assertEquals(Collections.singletonList("hi"), received, "成功路径应把 payload 投给业务"); } + + @Test + void consume_inProgress_reconsumesLaterWithoutLandingEvenIfExhausted() { + List landed = new ArrayList<>(); + MqSubscription sub = MqSubscription.builder() + .topic("t").group("g").payloadType(String.class).maxRetry(3) + .listener(m -> { throw new MqMessageInProgressException("k1"); }) + .retryExhaustedHandler(landed::add) + .build(); + + // reconsumeTimes=3 >= maxRetry=3:普通异常会走"耗尽落库+ACK",而 InProgress 是控制流信号, + // 必须仍 RECONSUME_LATER 且不落库——否则并发中的消息被当耗尽 ACK,静默丢消息。 + ConsumeConcurrentlyStatus status = + newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3); + + assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status, + "处理中信号应保留消息交 broker 重投,绝不 ACK"); + assertEquals(0, landed.size(), "处理中信号不是重试耗尽,不得回调 retryExhaustedHandler 落库"); + } + + @Test + void consume_inProgress_notExhausted_alsoReconsumesLater() { + List landed = new ArrayList<>(); + MqSubscription sub = MqSubscription.builder() + .topic("t").group("g").payloadType(String.class).maxRetry(3) + .listener(m -> { throw new MqMessageInProgressException("k1"); }) + .retryExhaustedHandler(landed::add) + .build(); + + ConsumeConcurrentlyStatus status = + newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 0)), 3); + + assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status); + assertEquals(0, landed.size()); + } }