完善死信队列

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