增加背压策略

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())
+                            );
                         }
                     }