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 投给业务"); + } +}