From 86661c687aca65838c680b0bf8f512778ab0f128 Mon Sep 17 00:00:00 2001
From: Cold <16419454+cold502@user.noreply.gitee.com>
Date: Sat, 6 Dec 2025 10:13:38 +0800
Subject: [PATCH] =?UTF-8?q?map=E9=A2=84=E5=88=86=E9=85=8D=E5=AE=B9?=
 =?UTF-8?q?=E9=87=8F=E9=81=BF=E5=85=8D=E5=8A=A8=E6=80=81=E6=89=A9=E5=AE=B9?=
 =?UTF-8?q?,=E4=BC=98=E5=8C=96=E4=BA=86=E8=BF=94=E5=9B=9E=E5=80=BC?=
 =?UTF-8?q?=E5=A4=8D=E7=94=A8=E6=83=85=E5=86=B5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 ragflow/client.go |  3 +--
 redis/redis.go    | 20 ++++++++++++--------
 redis/types.go    | 39 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 52 insertions(+), 10 deletions(-)
 create mode 100644 redis/types.go

diff --git a/ragflow/client.go b/ragflow/client.go
index 95ef503..0633dfc 100644
--- a/ragflow/client.go
+++ b/ragflow/client.go
@@ -50,8 +50,7 @@ func loadConfig(ctx context.Context) (baseURL, apiKey string) {
 	// 使用 GoFrame 全局配置（从项目的 config.yml 读取）
 	baseURL = g.Cfg().MustGet(ctx, "ragflow.base_url", "").String()
 	apiKey = g.Cfg().MustGet(ctx, "ragflow.api_key", "").String()
-
-	return baseURL, apiKey
+	return
 }
 
 // GetGlobalClient 获取全局客户端
diff --git a/redis/redis.go b/redis/redis.go
index ee17d3b..7856a4b 100644
--- a/redis/redis.go
+++ b/redis/redis.go
@@ -90,7 +90,8 @@ func ReadFromStream(ctx context.Context, streamKey, groupName, consumerName stri
 
 	// 解析返回值
 	// 格式: [[streamKey, [[msgID, [field1, value1, field2, value2, ...]], ...]]]
-	messages := []StreamMessage{}
+	// 预分配容量，避免动态扩容
+	messages := make([]StreamMessage, 0, int(count))
 
 	if result == nil {
 		// 超时或没有数据
@@ -130,8 +131,8 @@ func ReadFromStream(ctx context.Context, streamKey, groupName, consumerName stri
 				continue
 			}
 
-			// 解析字段为 map
-			values := make(map[string]interface{})
+			// 解析字段为 map，预分配容量,避免动态扩容
+			values := make(map[string]interface{}, len(fieldsArray)/2)
 			for i := 0; i < len(fieldsArray); i += 2 {
 				if i+1 < len(fieldsArray) {
 					key := gconv.String(fieldsArray[i])
@@ -154,7 +155,9 @@ func ReadFromStream(ctx context.Context, streamKey, groupName, consumerName stri
 // 使用 gredis Do() 方法执行 XACK 命令
 func AckMessage(ctx context.Context, streamKey, groupName string, messageIDs ...string) error {
 	// XACK streamKey groupName messageID1 messageID2 ...
-	args := []interface{}{streamKey, groupName}
+	// 预分配容量，避免动态扩容
+	args := make([]interface{}, 0, len(messageIDs)+2)
+	args = append(args, streamKey, groupName)
 	for _, id := range messageIDs {
 		args = append(args, id)
 	}
@@ -245,7 +248,8 @@ func ClaimPendingMessage(ctx context.Context, streamKey, groupName, consumerName
 		return []StreamMessage{}, nil
 	}
 
-	var messages []StreamMessage
+	// 预分配容量,避免动态扩容
+	messages := make([]StreamMessage, 0, len(messagesArray))
 	for _, msgData := range messagesArray {
 		msgArray, ok := msgData.([]interface{})
 		if !ok || len(msgArray) < 2 {
@@ -258,12 +262,12 @@ func ClaimPendingMessage(ctx context.Context, streamKey, groupName, consumerName
 			continue
 		}
 
-		values := make(map[string]interface{})
+		// 预分配 map 容量 ,避免动态扩容
+		values := make(map[string]interface{}, len(fieldsArray)/2)
 		for i := 0; i < len(fieldsArray); i += 2 {
 			if i+1 < len(fieldsArray) {
 				key := gconv.String(fieldsArray[i])
-				val := fieldsArray[i+1]
-				values[key] = val
+				values[key] = fieldsArray[i+1]
 			}
 		}
 
diff --git a/redis/types.go b/redis/types.go
new file mode 100644
index 0000000..8671a1a
--- /dev/null
+++ b/redis/types.go
@@ -0,0 +1,39 @@
+package redis
+
+// SendStreamMessage 发送到 Redis Stream 的消息结构
+type SendStreamMessage struct {
+	UserId    string `json:"user_id"`    // 用户ID
+	Content   string `json:"content"`    // 消息内容
+	Timestamp int64  `json:"timestamp"`  // 时间戳（秒）
+	MessageId string `json:"message_id"` // 消息唯一ID
+}
+
+// ToMap 转换为 map[string]interface{} 用于 Stream 存储
+func (m *SendStreamMessage) ToMap() map[string]interface{} {
+	return map[string]interface{}{
+		"user_id":    m.UserId,
+		"content":    m.Content,
+		"timestamp":  m.Timestamp,
+		"message_id": m.MessageId,
+	}
+}
+
+// BatchStreamMessage 批量消息结构
+type BatchStreamMessage struct {
+	UserId    string `json:"user_id"`   // 用户ID
+	Content   string `json:"content"`   // 消息内容
+	Timestamp int64  `json:"timestamp"` // 时间戳（秒）
+	BatchId   string `json:"batch_id"`  // 批次ID
+	Index     int    `json:"index"`     // 批次内序号
+}
+
+// ToMap 转换为 map[string]interface{} 用于 Stream 存储
+func (m *BatchStreamMessage) ToMap() map[string]interface{} {
+	return map[string]interface{}{
+		"user_id":   m.UserId,
+		"content":   m.Content,
+		"timestamp": m.Timestamp,
+		"batch_id":  m.BatchId,
+		"index":     m.Index,
+	}
+}