diff --git a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java index 3c4c22b..4367038 100644 --- a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java +++ b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java @@ -11,12 +11,13 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.stream.*; -import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.stereotype.Component; import java.net.InetAddress; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.List; import java.util.Map; @@ -37,6 +38,19 @@ public class RedisStreamConsumer { private static final Gson gson = new Gson(); + /** + * 普通业务 Stream 最大保留消息数。 + * Redis Stream 中，即使消息已经 ACK，也不会自动删除，如果不进行 Trim，Stream 会一直增长，最终占满 Redis 内存。 + */ + private static final long STREAM_MAX_LEN = 100_000L; //100_000L; + + /** + * 死信队列（DLQ）最大保留消息数。 + * DLQ 中保存的是处理失败的消息，一般数量远少于正常 Stream，因此可以设置得更小。 + */ + private static final long DLQ_MAX_LEN = 10_000L; + + private static final int PARTITIONS = 16; private static final String STREAM_PREFIX = "aeon_tcp_stream_"; private static final String GROUP = "aeon_tcp_stream_consumer_group"; @@ -48,7 +62,7 @@ public class RedisStreamConsumer { /** * 读取 Pending 消息和 reclaim 时 批量处理消息数 */ - private static final int BATCH_SIZE = 50; + private static final int BATCH_SIZE = 200; private volatile boolean running = true; private String consumerName; @@ -56,7 +70,7 @@ public class RedisStreamConsumer { /** * 接收消息总数 */ - private final LongAdder receiveCounter = new LongAdder(); + private final LongAdder processSubmitCounter = new LongAdder(); private final ExecutorService consumerExecutor = Executors.newFixedThreadPool(PARTITIONS, new CustomizableThreadFactory("redis-stream-consumer-")); @@ -90,6 +104,18 @@ public class RedisStreamConsumer { new CustomizableThreadFactory("redis-stats-") ); + /** + * 定时执行 Stream Trim 的线程池。 + * 单线程即可，因为： + * 1、Trim 操作频率很低（例如每 30 分钟一次） + * 2、每次只是循环处理 16 个 Stream + 1 个 DLQ + * 3、不需要并发执行 + */ + private final ScheduledExecutorService trimScheduler = + Executors.newSingleThreadScheduledExecutor( + new CustomizableThreadFactory("redis-trim-") + ); + @PostConstruct public void start() throws Exception { consumerName = buildConsumerName(); @@ -104,8 +130,6 @@ public class RedisStreamConsumer { int partition = i; consumerExecutor.submit(() -> { - // 启动时处理未 ack 消息 - recoverPending(stream); // 主消费循环 consumeLoop(stream, partition); }); @@ -121,9 +145,16 @@ public class RedisStreamConsumer { // 定时统计接收次数 statsScheduler.scheduleAtFixedRate(() -> { - long count = receiveCounter.sum(); + long count = processSubmitCounter.sum(); log.info("[F10] RedisStreamConsumer receive count={}", count); }, 10, 10, TimeUnit.SECONDS); + + trimScheduler.scheduleWithFixedDelay( + this::trimStreams, + 10, + 30, + TimeUnit.MINUTES + ); } private String buildConsumerName() throws Exception { @@ -163,8 +194,20 @@ public class RedisStreamConsumer { private void createGroup(String stream) { try { - redisTemplate.opsForStream() - .createGroup(stream, ReadOffset.from("0-0"), GROUP); +// redisTemplate.opsForStream() +// .createGroup(stream, ReadOffset.from("0-0"), GROUP); + redisTemplate.execute((RedisCallback) connection -> { + connection.execute( + "XGROUP", + "CREATE".getBytes(StandardCharsets.UTF_8), + stream.getBytes(StandardCharsets.UTF_8), + GROUP.getBytes(StandardCharsets.UTF_8), + "0-0".getBytes(StandardCharsets.UTF_8), + "MKSTREAM".getBytes(StandardCharsets.UTF_8) + ); + + return null; + }); log.info("create group success stream={}", stream); } catch (Exception e) { Throwable cause = e.getCause(); @@ -177,48 +220,6 @@ public class RedisStreamConsumer { } } - /** - * 启动时恢复未 ack 消息 - */ - private void recoverPending(String stream) { - try { - // 分批次处理 Pending - while (true) { - PendingMessages pending = redisTemplate.opsForStream() - .pending(stream, GROUP, Range.unbounded(), BATCH_SIZE); - - if (pending == null || pending.isEmpty()) break; - - for (PendingMessage msg : pending) { - // claim 未 ack 消息 - List> claimed = - redisTemplate.opsForStream().claim( - stream, - GROUP, - consumerName, - IDLE_TIMEOUT, - msg.getId() - ); - - for (MapRecord record : claimed) { - receiveCounter.increment(); -// workerExecutor.submit(() -> process(record, stream, msg.getTotalDeliveryCount())); - submitTask(() -> - process(record, stream, msg.getTotalDeliveryCount()) - ); - } - } - - // 如果少于 BATCH_SIZE，说明已处理完 - if (pending.size() < BATCH_SIZE) break; - } - - log.info("recoverPending stream={}", stream); - } catch (Exception e) { - log.error("recoverPending error stream={}", stream, e); - } - } - private void consumeLoop(String stream, int partition) { while (running) { try { @@ -233,7 +234,7 @@ public class RedisStreamConsumer { if (messages == null || messages.isEmpty()) continue; - receiveCounter.add(messages.size()); + processSubmitCounter.add(messages.size()); for (MapRecord message : messages) { // workerExecutor.submit(() -> process(message, stream, 1)); @@ -323,49 +324,108 @@ public class RedisStreamConsumer { */ private void reclaimIdleMessages() { if (workerExecutor.getQueue().size() > WORK_QUEUE_LIMIT) { - log.warn("skip reclaim, worker queue busy"); + log.warn("skip reclaim, worker queue busy, queueSize={}", workerExecutor.getQueue().size()); return; } for (int i = 1; i <= PARTITIONS; i++) { String stream = STREAM_PREFIX + i; try { - while (true) { - PendingMessages pending = - redisTemplate.opsForStream().pending(stream, GROUP, Range.unbounded(), BATCH_SIZE); + PendingMessages pending = + redisTemplate.opsForStream().pending(stream, GROUP, Range.unbounded(), BATCH_SIZE); + if (pending == null || pending.isEmpty()) { + continue; + } + for (PendingMessage p : pending) { + if (p.getElapsedTimeSinceLastDelivery().compareTo(IDLE_TIMEOUT) < 0) { + continue; + } + List> claimed = + redisTemplate.opsForStream() + .claim( + stream, + GROUP, + consumerName, + IDLE_TIMEOUT, + p.getId() + ); + if (claimed == null || claimed.isEmpty()) { + continue; + } + for(MapRecord r : claimed){ + processSubmitCounter.increment(); +// workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount())); + submitTask(() -> + process(r, stream, p.getTotalDeliveryCount()) + ); + } + } + } catch(Exception e){ + log.error("reclaim error stream={}",stream,e + ); + } + } + } - if (pending == null || pending.isEmpty()) break; + /** + * 定时裁剪所有 Stream。 + * 当前系统包括： + * + * 业务 Stream： + * aeon_tcp_stream_1 + * aeon_tcp_stream_2 + * ... + * aeon_tcp_stream_16 + * + * 死信队列： + * aeon_tcp_stream_dlq + * + * 主 Stream 和 DLQ 分开裁剪， + * 可以使用不同的最大保留数量。 + */ + private void trimStreams() { - for (PendingMessage p : pending) { - if (p.getElapsedTimeSinceLastDelivery().compareTo(IDLE_TIMEOUT) < 0) continue; + // 裁剪所有业务 Stream + for (int i = 1; i <= PARTITIONS; i++) { + trimStream(STREAM_PREFIX + i, STREAM_MAX_LEN); + } - List> claimed = - redisTemplate.opsForStream().claim( - stream, - GROUP, - consumerName, - IDLE_TIMEOUT, - p.getId() - ); + // 裁剪死信队列 + trimStream(DLQ_STREAM, DLQ_MAX_LEN); + } - for (MapRecord r : claimed) { - receiveCounter.increment(); -// workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount())); - submitTask(() -> - process(r, stream, p.getTotalDeliveryCount()) - ); - } - } + /** + * 裁剪指定 Stream。 + * Redis 命令：XTRIM stream MAXLEN ~ maxLen + * 例如：XTRIM aeon_tcp_stream_1 MAXLEN ~ 100000 表示：当 Stream 超过约10万条时，Redis 自动删除最旧的数据。 + * + * "~" 表示 Approximate Trim（近似裁剪）。 + * + * 为什么不用精确裁剪？ + * 精确裁剪：MAXLEN 1000000，Redis 每次都会精确删除到1000000条，CPU 消耗较高。 + * 近似裁剪：MAXLEN ~ 1000000，Redis 会在内部合适的时机进行裁剪，性能远高于精确裁剪。 + * + * @param streamName Stream 名称 + * @param maxLen 最大保留消息数量 + */ + private void trimStream(String streamName, long maxLen) { + try { + redisTemplate.execute((RedisCallback) connection -> { + connection.streamCommands().xTrim( + streamName.getBytes(StandardCharsets.UTF_8), + maxLen, + true + ); + return null; + }); - if (pending.size() < BATCH_SIZE) break; - } - } catch (Exception e) { - log.error("[F10] reclaim error stream={}", stream, e); - } + log.info("Trim stream success, stream={}, maxLen={}", streamName, maxLen); + } catch (Exception e) { + log.error("Trim stream failed, stream={}", streamName, e); } } public long getReceiveCount() { - return receiveCounter.sum(); + return processSubmitCounter.sum(); } private void sleep(long ms) { @@ -385,11 +445,13 @@ public class RedisStreamConsumer { workerExecutor.shutdown(); reclaimScheduler.shutdown(); statsScheduler.shutdown(); + trimScheduler.shutdown(); boolean consumerTerminated = consumerExecutor.awaitTermination(10, TimeUnit.SECONDS); boolean workerTerminated = workerExecutor.awaitTermination(10, TimeUnit.SECONDS); boolean reclaimTerminated = reclaimScheduler.awaitTermination(10, TimeUnit.SECONDS); boolean statsTerminated = statsScheduler.awaitTermination(10, TimeUnit.SECONDS); + boolean trimTerminated = trimScheduler.awaitTermination(10, TimeUnit.SECONDS ); if (!consumerTerminated) { consumerExecutor.shutdownNow(); @@ -403,11 +465,17 @@ public class RedisStreamConsumer { if (!statsTerminated) { statsScheduler.shutdownNow(); } + if (!trimTerminated) { + trimScheduler.shutdownNow(); + } - log.info("RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}", + log.info( + "RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}, trimTerminated={}", consumerTerminated, workerTerminated, reclaimTerminated, - statsTerminated); + statsTerminated, + trimTerminated + ); } } \ No newline at end of file