diff --git a/mq-starter-redis-stream/pom.xml b/mq-starter-redis-stream/pom.xml index 15f4a7e..60ad23c 100644 --- a/mq-starter-redis-stream/pom.xml +++ b/mq-starter-redis-stream/pom.xml @@ -3,4 +3,29 @@ 4.0.0 com.njcnmq-starter1.0.0-SNAPSHOT mq-starter-redis-stream + + com.njcnmq-starter-core1.0.0-SNAPSHOT + com.njcnredis-stream-springboot-starter1.0.0-SNAPSHOT + org.springframework.bootspring-boot-autoconfigure${autoconfigure.version} + com.alibabafastjson${fastjson.version} + org.projectlomboklombok${lombok.version}true + org.springframework.bootspring-boot-starter-test${springboot.version}test + org.springframework.bootspring-boot-starter-data-redis${springboot.version} + + + + org.apache.maven.pluginsmaven-compiler-plugin + + org.apache.maven.plugins + maven-surefire-plugin + + + **/*Test.java + **/*Tests.java + **/*IT.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 new file mode 100644 index 0000000..b4f09c6 --- /dev/null +++ b/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 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 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> 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 rec : recs) { + Map 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 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) 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 toStr(Map raw) { + Map 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(); } +} diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java new file mode 100644 index 0000000..65ea5b5 --- /dev/null +++ b/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); + } +} diff --git a/mq-starter-redis-stream/src/main/resources/META-INF/spring.factories b/mq-starter-redis-stream/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..cfc2394 --- /dev/null +++ b/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 diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverIT.java new file mode 100644 index 0000000..03ae0ea --- /dev/null +++ b/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 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; } + } +}