From aa24fb0f0295ed2b1008dc9f5ff2bc4d1de3c431 Mon Sep 17 00:00:00 2001
From: "review512jwy@163.com" <“review512jwy@163.com”>
Date: Mon, 13 Jul 2026 00:29:28 +0800
Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=AD=BB=E4=BF=A1=E9=98=9F?=
 =?UTF-8?q?=E5=88=97?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../listener/RedisStreamConsumer.java         | 19 ++++++++++++++-----
 1 file changed, 14 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 4367038..5330693 100644
--- a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
+++ b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
@@ -19,6 +19,7 @@ import org.springframework.stereotype.Component;
 import java.net.InetAddress;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -42,13 +43,13 @@ public class RedisStreamConsumer {
      * 普通业务 Stream 最大保留消息数。
      * Redis Stream 中，即使消息已经 ACK，也不会自动删除，如果不进行 Trim，Stream 会一直增长，最终占满 Redis 内存。
      */
-    private static final long STREAM_MAX_LEN = 100_000L; //100_000L;
+    private static final long STREAM_MAX_LEN = 1_000_000L; //100_000L;
 
     /**
      * 死信队列（DLQ）最大保留消息数。
      * DLQ 中保存的是处理失败的消息，一般数量远少于正常 Stream，因此可以设置得更小。
      */
-    private static final long DLQ_MAX_LEN = 10_000L;
+    private static final long DLQ_MAX_LEN = 100_000L;
 
 
     private static final int PARTITIONS = 16;
@@ -307,14 +308,22 @@ public class RedisStreamConsumer {
 
     private void moveToDlq(String stream, MapRecord<String, Object, Object> message) {
         try {
+            // 必须新建一个 Map，不要直接修改原对象
+            Map<Object, Object> dlqBody = new HashMap<>(message.getValue());
+
+            // 注入排障上下文
+            dlqBody.put("_dlq_source_stream", stream);
+            dlqBody.put("_dlq_original_id", message.getId().getValue());
+            dlqBody.put("_dlq_timestamp", String.valueOf(System.currentTimeMillis()));
+
             redisTemplate.opsForStream().add(
-                    StreamRecords.mapBacked(message.getValue())
+                    StreamRecords.mapBacked(dlqBody)
                             .withStreamKey(DLQ_STREAM)
             );
             ack(stream, message);
-            log.error("[F10] move to DLQ :{}", message);
+            log.error("[F10] move to DLQ : stream={}, id={}", stream, message.getId());
         } catch (Exception e) {
-            log.error("[F10] dlq error :{}", message, e);
+            log.error("[F10] dlq error : stream={}, id={}", stream, message.getId(), e);
         }
     }