feat(mq-starter-redis-stream): redis driver 复用 Phase1 低层件 + 薄收发循环

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-redis-stream/pom.xml

@@ -3,4 +3,29 @@
     <modelVersion>4.0.0</modelVersion>
     <parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
     <artifactId>mq-starter-redis-stream</artifactId>
+    <dependencies>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>com.njcn</groupId><artifactId>redis-stream-springboot-starter</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-autoconfigure</artifactId><version>${autoconfigure.version}</version></dependency>
+        <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
+        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>${springboot.version}</version></dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId></plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <includes>
+                        <include>**/*Test.java</include>
+                        <include>**/*Tests.java</include>
+                        <include>**/*IT.java</include>
+                    </includes>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -0,0 +1,116 @@
+package com.njcn.mq.driver.redis;
+import com.alibaba.fastjson.JSON;
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.codec.MessageCodec;
+import com.njcn.middle.stream.constant.StreamMessageConstant;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqSubscription;
+import lombok.extern.slf4j.Slf4j;
+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.*;
+
+@Slf4j
+public class RedisStreamMqDriver implements MqDriver {
+    private final StringRedisTemplate redis;
+    private final StreamKeyBuilder keyBuilder;
+    private final RedisStreamProperties props;
+    private final List<Thread> workers = new ArrayList<>();
+    private volatile boolean running = true;
+
+    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
+        this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
+    }
+
+    @Override public void send(String topic, MqMessage message) {
+        String stream = keyBuilder.buildStream(topic);
+        String json = JSON.toJSONString(message.getPayload());
+        MessageCodec.Encoded enc = MessageCodec.encodeBody(json, false); // 下行恒 plain
+        Map<String, String> f = new HashMap<>(8);
+        f.put(StreamMessageConstant.F_KEY, message.getKey());
+        f.put(StreamMessageConstant.F_TAG, message.getTag());
+        f.put(StreamMessageConstant.F_ENC, enc.getEnc());
+        f.put(StreamMessageConstant.F_BODY, enc.getBody());
+        f.put(StreamMessageConstant.F_TS, String.valueOf(System.currentTimeMillis()));
+        redis.opsForStream().add(StreamRecords.newRecord().in(stream).ofMap(f));
+    }
+
+    @Override public void subscribe(MqSubscription sub) {
+        String stream = keyBuilder.buildStream(sub.getTopic());
+        String group = sub.getGroup();
+        ensureGroup(stream, group);
+        String consumer = (props.getConsumerName() == null || props.getConsumerName().isEmpty())
+                ? "mq-" + UUID.randomUUID() : props.getConsumerName();
+        Thread t = new Thread(() -> loop(sub, stream, group, consumer), "mq-redis-" + stream + "-" + group);
+        t.setDaemon(true); workers.add(t); t.start();
+    }
+
+    private void loop(MqSubscription sub, String stream, String group, String consumer) {
+        int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : props.getMaxRetry();
+        while (running && !Thread.currentThread().isInterrupted()) {
+            try {
+                List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
+                        Consumer.from(group, consumer),
+                        StreamReadOptions.empty().count(props.getReadCount()).block(Duration.ofMillis(props.getBlockMs())),
+                        StreamOffset.create(stream, ReadOffset.lastConsumed()));
+                if (recs == null || recs.isEmpty()) continue;
+                for (MapRecord<String, Object, Object> rec : recs) {
+                    Map<String, String> fields = toStr(rec.getValue());
+                    boolean ack = dispatch(sub, fields, maxRetry);
+                    if (ack) redis.opsForStream().acknowledge(stream, group, rec.getId());
+                }
+            } catch (Exception e) {
+                if (!running) break;
+                log.error("[mq-redis] loop error stream={} group={}", stream, group, e);
+                sleep(props.getBlockMs());
+            }
+        }
+    }
+
+    /** @return true=ACK；false=留 PEL（重试）。 */
+    private boolean dispatch(MqSubscription sub, Map<String, String> fields, int maxRetry) {
+        Object payload;
+        try {
+            String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
+            payload = JSON.parseObject(json, sub.getPayloadType());
+        } catch (Exception poison) {
+            log.error("[mq-redis] poison message acked, key={}", fields.get(StreamMessageConstant.F_KEY), poison);
+            return true; // 毒消息直接 ACK，不卡队列
+        }
+        MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
+        try {
+            sub.getListener().onMessage(msg);
+            return true;
+        } catch (Exception e) {
+            log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e);
+            return false; // 留 PEL；超 maxRetry 的兜底由后续 autoclaim 增强补（轻量版先靠重投）
+        }
+    }
+
+    private void ensureGroup(String stream, String group) {
+        try {
+            redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",
+                    "CREATE".getBytes(StandardCharsets.UTF_8), stream.getBytes(StandardCharsets.UTF_8),
+                    group.getBytes(StandardCharsets.UTF_8), "$".getBytes(StandardCharsets.UTF_8),
+                    "MKSTREAM".getBytes(StandardCharsets.UTF_8)));
+        } catch (RuntimeException e) {
+            for (Throwable x = e; x != null; x = x.getCause())
+                if (x.getMessage() != null && x.getMessage().contains("BUSYGROUP")) return;
+            throw e;
+        }
+    }
+
+    private static Map<String, String> toStr(Map<Object, Object> raw) {
+        Map<String, String> m = new HashMap<>(Math.max(8, raw.size() * 2));
+        raw.forEach((k, v) -> m.put(String.valueOf(k), v == null ? null : String.valueOf(v)));
+        return m;
+    }
+    private static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } }
+
+    @Override public void stop() { running = false; for (Thread t : workers) t.interrupt(); workers.clear(); }
+}

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java

@@ -0,0 +1,21 @@
+package com.njcn.mq.driver.redis;
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.mq.spi.MqDriver;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+@Configuration
+@ConditionalOnClass(StringRedisTemplate.class)
+@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream")
+public class RedisStreamMqDriverAutoConfiguration {
+    @Bean
+    @ConditionalOnMissingBean
+    public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) {
+        return new RedisStreamMqDriver(redis, streamKeyBuilder, props);
+    }
+}

mq-starter-redis-stream/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverIT.java

@@ -0,0 +1,47 @@
+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.core.StringRedisTemplate;
+import java.util.concurrent.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisStreamMqDriverIT {
+    @Test
+    void sendThenSubscribe_roundTrips() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");   // 不开 env 隔离，stream 名=topic
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-consumer");
+        props.setReadCount(10);
+        props.setBlockMs(2000);
+        props.setMaxRetry(3);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props);
+
+        String topic = "MQ_IT_Topic_" + System.nanoTime();
+        BlockingQueue<String> got = new LinkedBlockingQueue<>();
+        driver.subscribe(MqSubscription.builder()
+                .topic(topic).group("g1").tag("*").payloadType(String.class)
+                .concurrency(1).batchSize(1).maxRetry(3)
+                .listener(m -> got.add((String) m.getPayload())).build());
+
+        driver.send(topic, MqMessage.of("k1", "", "hello-redis"));
+        assertEquals("hello-redis", got.poll(5, TimeUnit.SECONDS));
+        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; }
+    }
+}