From 83656215f6af90ca7be5e47af196c8852286d984 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 16 Jul 2026 13:23:41 +0800 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-redis-stream):=20=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=E7=AB=AF=E6=8C=89=E8=AE=A2=E9=98=85=20tag=20=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=EF=BC=8C=E4=B8=8D=E5=8C=B9=E9=85=8D=20ACK=20=E4=B8=A2?= =?UTF-8?q?=E5=BC=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dispatch 顶部（反序列化前）经 MqTagMatcher 比对，不匹配直接 ACK： 不碰去重、不攒批、不重试；覆盖正常消费与 reclaim 重投两条路径。 Co-Authored-By: Claude Fable 5 --- .../mq/driver/redis/RedisStreamMqDriver.java | 9 +++ .../redis/RedisStreamMqDriverTagFilterIT.java | 64 +++++++++++++++++++ 2 files changed, 73 insertions(+) create mode 100644 mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java 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 b071b0d..db0b21f 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 @@ -8,6 +8,7 @@ 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 com.njcn.mq.spi.MqTagMatcher; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.connection.stream.*; import org.springframework.data.domain.Range; @@ -87,6 +88,14 @@ public class RedisStreamMqDriver implements MqDriver { /** @return true=ACK；false=留 PEL（重试）。 */ private boolean dispatch(MqSubscription sub, Map fields, int maxRetry) { + String msgTag = fields.get(StreamMessageConstant.F_TAG); + if (!MqTagMatcher.matches(sub.getTag(), msgTag)) { + // tag 不匹配：不归本订阅消费，直接 ACK 丢弃（对齐 rocketmq broker 过滤语义，只是位置在消费端）。 + // 置于反序列化之前：不碰去重标记、不进攒批、不走重试，连 JSON 都不用解。 + log.debug("[mq-redis] tag 不匹配，ACK 跳过 key={} msgTag={} subTag={}", + fields.get(StreamMessageConstant.F_KEY), msgTag, sub.getTag()); + return true; + } Object payload; try { String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY)); 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 new file mode 100644 index 0000000..a428586 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java @@ -0,0 +1,64 @@ +package com.njcn.mq.driver.redis; + +import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqSubscription; +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.core.StringRedisTemplate; + +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.assertNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +class RedisStreamMqDriverTagFilterIT { + + @Test + void tagFilter_onlyMatchingDelivered_othersAckedCleanly() throws Exception { + StringRedisTemplate redis = newRedisOrSkip(); + StreamKeyBuilder kb = new StreamKeyBuilder(false, ""); + RedisStreamProperties props = new RedisStreamProperties(); + props.setConsumerName("it-tag"); + props.setReadCount(10); + props.setBlockMs(300); + props.setMaxRetry(3); + RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L); + + String topic = "MQ_TAG_IT_" + System.nanoTime(); + BlockingQueue got = new LinkedBlockingQueue<>(); + driver.subscribe(MqSubscription.builder() + .topic(topic).group("g1").tag("A").payloadType(String.class) + .concurrency(1).batchSize(1).maxRetry(3) + .listener(m -> got.add((String) m.getPayload())) + .build()); + + driver.send(topic, MqMessage.of("k-a", "A", "match")); + driver.send(topic, MqMessage.of("k-b", "B", "mismatch")); + driver.send(topic, MqMessage.of("k-none", "", "untagged")); + + assertEquals("match", got.poll(5, TimeUnit.SECONDS), "tag=A 的消息应正常消费"); + assertNull(got.poll(1500, TimeUnit.MILLISECONDS), "tag 不匹配/未打 tag 的消息不得进业务方法"); + + Thread.sleep(500); // 留时间让过滤 ACK 落地 + PendingMessagesSummary summary = redis.opsForStream().pending(kb.buildStream(topic), "g1"); + assertEquals(0L, summary.getTotalPendingMessages(), "被过滤的消息应 ACK 干净，PEL 无残留"); + driver.stop(); + } + + private static StringRedisTemplate newRedisOrSkip() { + try { + LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379); + cf.afterPropertiesSet(); + StringRedisTemplate t = new StringRedisTemplate(cf); + t.afterPropertiesSet(); + t.hasKey("ping"); // 触发真实连接 + return t; + } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; } + } +}