feat: HTTP模式支持 - 优化Stream消费逻辑，添加pending消息优先读取

redis/redis.go

@@ -190,9 +190,6 @@ func CreateConsumerGroup(ctx context.Context, streamKey, groupName string) error
 // ReadFromStream 从 Stream 读取消息（消费者组模式）
 // 使用 gredis Do() 方法执行 XREADGROUP 命令
 func ReadFromStream(ctx context.Context, streamKey, groupName, consumerName string, count int64, blockMs int64) ([]StreamMessage, error) {
-	glog.Debugf(ctx, "[DEBUG Redis] XREADGROUP GROUP %s %s COUNT %d BLOCK %d STREAMS %s >",
-		groupName, consumerName, count, blockMs, streamKey)
-
 	// 检查是否需要记录trace（避免轮询产生大量trace）
 	execCtx := ctx
 	if !g.Cfg().MustGet(ctx, "jaeger.traceStream", true).Bool() {
@@ -201,17 +198,36 @@ func ReadFromStream(ctx context.Context, streamKey, groupName, consumerName stri
 	}
 
 RECONNECT:
-	// XREADGROUP GROUP groupName consumerName COUNT count BLOCK blockMs STREAMS streamKey >
+	// 先尝试读取pending消息（ID=0），处理积压
+	glog.Debugf(ctx, "[DEBUG Redis] XREADGROUP GROUP %s %s COUNT %d BLOCK 0 STREAMS %s 0",
+		groupName, consumerName, count, streamKey)
+
 	result, err := redisClient.Do(execCtx,
 		"XREADGROUP", "GROUP", groupName, consumerName,
 		"COUNT", count,
-		"BLOCK", blockMs,
-		"STREAMS", streamKey, ">",
+		"BLOCK", 0, // 不阻塞，立即返回
+		"STREAMS", streamKey, "0", // ID=0 读取pending消息
 	)
 	if err != nil {
 		goto RECONNECT
 	}
 
+	// 如果没有pending消息，读取新消息
+	if result == nil || result.IsEmpty() {
+		glog.Debugf(ctx, "[DEBUG Redis] 无pending消息，读取新消息 XREADGROUP GROUP %s %s COUNT %d BLOCK %d STREAMS %s >",
+			groupName, consumerName, count, blockMs, streamKey)
+
+		result, err = redisClient.Do(execCtx,
+			"XREADGROUP", "GROUP", groupName, consumerName,
+			"COUNT", count,
+			"BLOCK", blockMs,
+			"STREAMS", streamKey, ">",
+		)
+		if err != nil {
+			goto RECONNECT
+		}
+	}
+
 	glog.Debugf(ctx, "[DEBUG Redis] XREADGROUP 返回: %+v", result)
 
 	// 预分配容量，避免动态扩容