From 1f7338678dafe61e0957aa2f78342dc4987ec1eb Mon Sep 17 00:00:00 2001
From: "review512jwy@163.com" <“review512jwy@163.com”>
Date: Sun, 12 Jul 2026 11:50:35 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E8=83=8C=E5=8E=8B=E7=AD=96?=
 =?UTF-8?q?=E7=95=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../listener/RedisStreamConsumer.java         | 39 ++++++++++++++++---
 1 file changed, 34 insertions(+), 5 deletions(-)

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 6bf11ba..3c4c22b 100644
--- a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
+++ b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
@@ -71,17 +71,24 @@ public class RedisStreamConsumer {
                     new ThreadPoolExecutor.CallerRunsPolicy()
             );
 
+    // new LinkedBlockingQueue<>(10000)的80%
+    private static final int WORK_QUEUE_LIMIT = 8000;
+
     /**
      * 防止消息丢失或长时间未处理（死消息）
      */
     private final ScheduledExecutorService reclaimScheduler =
-            Executors.newSingleThreadScheduledExecutor();
+            Executors.newSingleThreadScheduledExecutor(
+                    new CustomizableThreadFactory("redis-reclaim-")
+            );
 
     /**
      * 统计接收次数任务
      */
     private final ScheduledExecutorService statsScheduler =
-            Executors.newSingleThreadScheduledExecutor();
+            Executors.newSingleThreadScheduledExecutor(
+                    new CustomizableThreadFactory("redis-stats-")
+            );
 
     @PostConstruct
     public void start() throws Exception {
@@ -195,7 +202,10 @@ public class RedisStreamConsumer {
 
                     for (MapRecord<String, Object, Object> record : claimed) {
                         receiveCounter.increment();
-                        workerExecutor.submit(() -> process(record, stream, msg.getTotalDeliveryCount()));
+//                        workerExecutor.submit(() -> process(record, stream, msg.getTotalDeliveryCount()));
+                        submitTask(() ->
+                                process(record, stream, msg.getTotalDeliveryCount())
+                        );
                     }
                 }
 
@@ -226,7 +236,10 @@ public class RedisStreamConsumer {
                 receiveCounter.add(messages.size());
 
                 for (MapRecord<String, Object, Object> message : messages) {
-                    workerExecutor.submit(() -> process(message, stream, 1));
+//                    workerExecutor.submit(() -> process(message, stream, 1));
+                    submitTask(() ->
+                            process(message, stream, 1)
+                    );
                 }
 
             } catch (Exception e) {
@@ -236,6 +249,15 @@ public class RedisStreamConsumer {
         }
     }
 
+    private void submitTask(Runnable task) {
+        while (running && workerExecutor.getQueue().size() > WORK_QUEUE_LIMIT) {
+            sleep(100);
+        }
+        if (running) {
+            workerExecutor.submit(task);
+        }
+    }
+
     private void process(MapRecord<String, Object, Object> message,
                          String stream,
                          long deliveryCount) {
@@ -300,6 +322,10 @@ public class RedisStreamConsumer {
      * 防止消息被遗忘或丢失
      */
     private void reclaimIdleMessages() {
+        if (workerExecutor.getQueue().size() > WORK_QUEUE_LIMIT) {
+            log.warn("skip reclaim, worker queue busy");
+            return;
+        }
         for (int i = 1; i <= PARTITIONS; i++) {
             String stream = STREAM_PREFIX + i;
             try {
@@ -323,7 +349,10 @@ public class RedisStreamConsumer {
 
                         for (MapRecord<String, Object, Object> r : claimed) {
                             receiveCounter.increment();
-                            workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount()));
+//                            workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount()));
+                            submitTask(() ->
+                                    process(r, stream, p.getTotalDeliveryCount())
+                            );
                         }
                     }