From 29473e6fef2532b3105e3fdc1978e565c122159d Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 9 Jul 2026 20:41:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-rocketmq):=20=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E8=80=97=E5=B0=BD=E5=9B=9E=E8=B0=83=20retryExhaustedHandler=20?= =?UTF-8?q?=E8=90=BD=E5=BA=93=E5=B9=B6=20ACK=EF=BC=8CmaxRetry=20=E5=AF=B9?= =?UTF-8?q?=E9=BD=90=20@MqListener?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对齐 redis-stream 的 reclaim 耗尽语义，兑现"换 MQ 业务不动"契约：此前 rocketmq 只有 broker 默认 16 次重投、无 RETRY_EXHAUSTED 落库。 - setMaxReconsumeTimes 对齐 @MqListener(maxRetry)（缺省 3），不再用 broker 默认 16 - 抽出 consumeBatch：reconsumeTimes 达 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止；未达则 RECONSUME_LATER - onRetryExhausted 落库 best-effort：毒消息 payload 反序列化失败时留 null、至少保留 key/tag，并隔离 handler 自身异常 - 补消费逻辑单测（耗尽落库 / 未耗尽重投 / 成功投递） Co-Authored-By: Claude Opus 4.8 (1M context) --- .../mq/driver/rocketmq/RocketMqDriver.java | 56 +++++++++--- .../rocketmq/RocketMqDriverConsumeTest.java | 87 +++++++++++++++++++ 2 files changed, 130 insertions(+), 13 deletions(-) create mode 100644 mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java 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 3a4b43e..6d40765 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 @@ -16,6 +16,7 @@ import org.springframework.messaging.support.MessageBuilder; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; @Slf4j public class RocketMqDriver implements MqDriver { @@ -41,23 +42,14 @@ public class RocketMqDriver implements MqDriver { @Override public void subscribe(MqSubscription sub) { try { + int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : 3; DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(sub.getGroup()); consumer.setNamesrvAddr(namesrvAddr); + consumer.setMaxReconsumeTimes(maxRetry); // 对齐 @MqListener(maxRetry),而非用 broker 默认 16 次 String tag = sub.getTag(); consumer.subscribe(sub.getTopic(), (tag == null || tag.isEmpty()) ? "*" : tag); - consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { - for (MessageExt ext : msgs) { - try { - 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 (Exception e) { - log.warn("[mq-rocket] handle failed msgId={}", ext.getMsgId(), e); - return ConsumeConcurrentlyStatus.RECONSUME_LATER; - } - } - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - }); + consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> + consumeBatch(sub, msgs, maxRetry)); consumer.start(); consumers.add(consumer); } catch (Exception e) { @@ -65,6 +57,44 @@ public class RocketMqDriver implements MqDriver { } } + ConsumeConcurrentlyStatus consumeBatch(MqSubscription sub, List msgs, int maxRetry) { + for (MessageExt ext : msgs) { + try { + 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 (Exception e) { + if (ext.getReconsumeTimes() >= maxRetry) { + // 重试耗尽:回调 retryExhaustedHandler 落库并 ACK 终止,不再无限重投 + //（对齐 redis-stream reclaim 耗尽语义,兑现"换 MQ 业务不动"契约)。 + onRetryExhausted(sub, ext, e); + } else { + log.warn("[mq-rocket] handle failed msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } + } + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + + /** 重试耗尽落库(best-effort:毒消息时 payload 为 null，至少保留 key/tag),并隔离 handler 自身异常。 */ + private void onRetryExhausted(MqSubscription sub, MessageExt ext, Exception cause) { + log.warn("[mq-rocket] retry exhausted, acked msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause); + Consumer handler = sub.getRetryExhaustedHandler(); + if (handler == null) return; + Object payload = null; + try { + payload = JSON.parseObject(new String(ext.getBody(), StandardCharsets.UTF_8), sub.getPayloadType()); + } catch (Exception ignore) { + // 毒消息:payload 留 null，至少保留 key/tag 供落库(对齐 redis buildMessage) + } + try { + handler.accept(MqMessage.of(ext.getKeys(), ext.getTags(), payload)); + } catch (Exception ex) { + log.error("[mq-rocket] retryExhausted handler error msgId={}（不影响主流程)", ext.getMsgId(), ex); + } + } + @Override public void stop() { for (DefaultMQPushConsumer c : consumers) c.shutdown(); 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 new file mode 100644 index 0000000..0f0fcad --- /dev/null +++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java @@ -0,0 +1,87 @@ +package com.njcn.mq.driver.rocketmq; + +import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqSubscription; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.mock; + +/** + * RocketMQ 消费处理逻辑单测(纯 POJO，不启动真实 consumer / 不连 namesrv）。 + * + *

守护跨 driver 契约一致性:与 redis-stream 的 reclaim 耗尽语义对齐 —— + * 重试次数达到 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止;未达到则 RECONSUME_LATER 交 broker 重投。 + */ +class RocketMqDriverConsumeTest { + + private RocketMqDriver newDriver() { + return new RocketMqDriver(mock(RocketMQTemplate.class), "localhost:9876"); + } + + private static MessageExt msg(String key, String body, int reconsumeTimes) { + MessageExt ext = new MessageExt(); + ext.setKeys(key); + ext.setBody(body.getBytes(StandardCharsets.UTF_8)); + ext.setReconsumeTimes(reconsumeTimes); + return ext; + } + + @Test + void consume_retryExhausted_callsHandlerAndAcks() { + List landed = new ArrayList<>(); + MqSubscription sub = MqSubscription.builder() + .topic("t").group("g").payloadType(String.class).maxRetry(3) + .listener(m -> { throw new RuntimeException("biz boom"); }) + .retryExhaustedHandler(landed::add) + .build(); + + ConsumeConcurrentlyStatus status = + newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3); + + assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status, + "重试耗尽应 ACK 终止,不再无限重投"); + assertEquals(1, landed.size(), "重试耗尽应回调 retryExhaustedHandler 落库(对齐 redis)"); + assertEquals("k1", landed.get(0).getKey(), "落库消息应保留原 key"); + } + + @Test + void consume_notExhausted_reconsumesLaterWithoutLanding() { + List landed = new ArrayList<>(); + MqSubscription sub = MqSubscription.builder() + .topic("t").group("g").payloadType(String.class).maxRetry(3) + .listener(m -> { throw new RuntimeException("biz boom"); }) + .retryExhaustedHandler(landed::add) + .build(); + + ConsumeConcurrentlyStatus status = + newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 1)), 3); + + assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status, + "未达 maxRetry 应交 broker 重投"); + assertEquals(0, landed.size(), "未耗尽不应落库"); + } + + @Test + void consume_success_deliversPayloadAndAcks() { + List received = new ArrayList<>(); + MqSubscription sub = MqSubscription.builder() + .topic("t").group("g").payloadType(String.class).maxRetry(3) + .listener(m -> received.add(m.getPayload())) + .build(); + + ConsumeConcurrentlyStatus status = + newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hi\"", 0)), 3); + + assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status); + assertEquals(Collections.singletonList("hi"), received, "成功路径应把 payload 投给业务"); + } +}