feat(mq-starter-redis-stream): 消费端按订阅 tag 过滤，不匹配 ACK 丢弃

dispatch 顶部（反序列化前）经 MqTagMatcher 比对，不匹配直接 ACK：
不碰去重、不攒批、不重试；覆盖正常消费与 reclaim 重投两条路径。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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<String, String> 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));

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<String> 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; }
+    }
+}