diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java index db0b21f..ab918f6 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java @@ -146,6 +146,13 @@ public class RedisStreamMqDriver implements MqDriver { Map fields = new HashMap<>(); rec.getValue().forEach((k, v) -> fields.put( new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8))); + if (!MqTagMatcher.matches(sub.getTag(), fields.get(StreamMessageConstant.F_TAG))) { + // tag 不匹配:不归本组消费,直接 ACK 丢弃——不得进耗尽分支以本组名义落库,也无须重投。 + redis.opsForStream().acknowledge(stream, group, rec.getId()); + log.debug("[mq-redis] tag 不匹配(reclaim),ACK 丢弃 key={} delivery={}", + fields.get(StreamMessageConstant.F_KEY), deliveryCount); + continue; + } if (deliveryCount > maxRetry) { if (sub.getRetryExhaustedHandler() != null) { try { diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java index a428586..c248081 100644 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java @@ -6,15 +6,21 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties; import com.njcn.middle.stream.support.StreamKeyBuilder; import org.junit.jupiter.api.Test; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; -import org.springframework.data.redis.connection.stream.PendingMessagesSummary; +import org.springframework.data.redis.connection.stream.*; +import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.StringRedisTemplate; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assumptions.assumeTrue; class RedisStreamMqDriverTagFilterIT { @@ -51,6 +57,59 @@ class RedisStreamMqDriverTagFilterIT { driver.stop(); } + @Test + void reclaimExhausted_mismatchedTag_ackedWithoutExhaustedCallback() throws Exception { + StringRedisTemplate redis = newRedisOrSkip(); + StreamKeyBuilder kb = new StreamKeyBuilder(false, ""); + RedisStreamProperties props = new RedisStreamProperties(); + props.setConsumerName("it-tag-reclaim"); + props.setReadCount(10); + props.setBlockMs(300); + props.setMaxRetry(1); + RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L); + + String topic = "MQ_TAG_RECLAIM_IT_" + System.nanoTime(); + String stream = kb.buildStream(topic); + + // 1. 手工建组(driver 尚未 subscribe,模拟历史遗留 PEL 场景) + redis.execute((RedisCallback) c -> c.execute("XGROUP", + "CREATE".getBytes(StandardCharsets.UTF_8), stream.getBytes(StandardCharsets.UTF_8), + "g1".getBytes(StandardCharsets.UTF_8), "$".getBytes(StandardCharsets.UTF_8), + "MKSTREAM".getBytes(StandardCharsets.UTF_8))); + + // 2. 发一条 tag=B 的消息(send 不依赖 subscribe) + driver.send(topic, MqMessage.of("k-stale", "B", "stale")); + + // 3. 手工 XREADGROUP 不 ACK:delivery=1,模拟消费者读走后崩溃 + List> recs = redis.opsForStream().read( + Consumer.from("g1", "seed"), + StreamReadOptions.empty().count(1), + StreamOffset.create(stream, ReadOffset.lastConsumed())); + assertNotNull(recs); + assertEquals(1, recs.size(), "种子消息应已进 PEL"); + RecordId staleId = recs.get(0).getId(); + + // 4. 手工 XCLAIM 一次:delivery=2 > maxRetry=1,reclaim 将走耗尽分支 + redis.execute((RedisCallback) conn -> conn.streamCommands() + .xClaim(stream.getBytes(StandardCharsets.UTF_8), "g1", "seed2", Duration.ZERO, staleId)); + + // 5. 启动订阅 tag=A 的 driver,等 reclaim 处理这条 tag=B 的残留 + BlockingQueue exhausted = new LinkedBlockingQueue<>(); + BlockingQueue delivered = new LinkedBlockingQueue<>(); + driver.subscribe(MqSubscription.builder() + .topic(topic).group("g1").tag("A").payloadType(String.class) + .concurrency(1).batchSize(1).maxRetry(1) + .listener(m -> delivered.add(m.getPayload())) + .retryExhaustedHandler(exhausted::add) + .build()); + + assertNull(exhausted.poll(3, TimeUnit.SECONDS), "tag 不匹配的消息不得以本组名义耗尽落库"); + assertTrue(delivered.isEmpty(), "tag 不匹配的消息不得进业务方法"); + PendingMessagesSummary summary = redis.opsForStream().pending(stream, "g1"); + assertEquals(0L, summary.getTotalPendingMessages(), "应被 ACK 丢弃,PEL 清空"); + driver.stop(); + } + private static StringRedisTemplate newRedisOrSkip() { try { LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);