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<MqMessage> 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<MqMessage> 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());
+    }
 }