From b693a733ba9149c0c51845acbb8e8af0e88296db Mon Sep 17 00:00:00 2001 From: fengsilin Date: Fri, 10 Apr 2026 15:56:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(region-sync):=20=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E7=94=A8=E6=88=B7=E4=BD=99=E9=A2=9D=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=8C=E4=BC=98=E5=8C=96=E5=90=8C=E6=AD=A5=E6=80=A7?= =?UTF-8?q?=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunQuotaSync 从逐个查询改为批量查询(每批500个), 减少 Master 节点 API 调用次数,提升同步效率。 新增: - BatchQueryUserQuota 控制器接口 - BatchQueryQuota 客户端方法 - QuotaEntry/BatchQueryQuotaRequest/Response 类型 附带 update-image.sh 部署脚本。 Co-Authored-By: Claude Opus 4.6 --- controller/region_sync.go | 35 +++++++++++++++++ service/region_sync/sync_client.go | 12 ++++++ service/region_sync/sync_manager.go | 44 ++++++++++++++------- service/region_sync/sync_types.go | 18 +++++++++ update-image.sh | 59 +++++++++++++++++++++++++++++ 5 files changed, 155 insertions(+), 13 deletions(-) create mode 100644 update-image.sh diff --git a/controller/region_sync.go b/controller/region_sync.go index 80bd904..cad5415 100644 --- a/controller/region_sync.go +++ b/controller/region_sync.go @@ -145,6 +145,41 @@ func QueryUserQuota(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"success": true, "quota": user.Quota}) } +// BatchQueryUserQuota 批量查询用户余额(供 slave 节点调用) +func BatchQueryUserQuota(c *gin.Context) { + var req region_sync.BatchQueryQuotaRequest + if err := c.ShouldBindJSON(&req); err != nil { + syncErrorResponse(c, http.StatusBadRequest, "invalid request body") + return + } + + if len(req.UserIds) == 0 { + c.JSON(http.StatusOK, region_sync.BatchQueryQuotaResponse{Success: true, Quotas: []region_sync.QuotaEntry{}}) + return + } + + if len(req.UserIds) > 500 { + syncErrorResponse(c, http.StatusBadRequest, "too many user ids (max 500)") + return + } + + var users []model.User + if err := model.DB.Where("id IN ?", req.UserIds).Find(&users).Error; err != nil { + syncErrorResponse(c, http.StatusInternalServerError, "database error") + return + } + + quotas := make([]region_sync.QuotaEntry, 0, len(users)) + for _, user := range users { + quotas = append(quotas, region_sync.QuotaEntry{UserId: user.Id, Quota: user.Quota}) + } + + c.JSON(http.StatusOK, region_sync.BatchQueryQuotaResponse{ + Success: true, + Quotas: quotas, + }) +} + // BatchDeductQuota 批量扣费(供 slave 节点调用 master 节点) func BatchDeductQuota(c *gin.Context) { var req region_sync.BatchDeductRequest diff --git a/service/region_sync/sync_client.go b/service/region_sync/sync_client.go index 8af24ce..cbd8c42 100644 --- a/service/region_sync/sync_client.go +++ b/service/region_sync/sync_client.go @@ -142,3 +142,15 @@ func (c *SyncClient) FetchConfig() (*SyncConfigResponse, error) { } return &resp, nil } + +func (c *SyncClient) BatchQueryQuota(userIds []int) (*BatchQueryQuotaResponse, error) { + data, err := c.doRequest("POST", "/api/internal/sync/quota/batch-query", &BatchQueryQuotaRequest{UserIds: userIds}) + if err != nil { + return nil, err + } + var resp BatchQueryQuotaResponse + if err := json.Unmarshal(data, &resp); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + return &resp, nil +} diff --git a/service/region_sync/sync_manager.go b/service/region_sync/sync_manager.go index b2e6ae9..341c101 100644 --- a/service/region_sync/sync_manager.go +++ b/service/region_sync/sync_manager.go @@ -143,7 +143,7 @@ func (m *SyncManager) QueryMasterQuota(remoteUserId int) (int, error) { return resp.Quota, nil } -// RunQuotaSync 从 Master 拉取所有同步用户的最新余额,更新本地 synced_quota +// RunQuotaSync 从 Master 批量拉取所有同步用户的最新余额,更新本地 synced_quota func (m *SyncManager) RunQuotaSync() int { settings := system_setting.GetRegionSyncSettings() if !settings.Enabled || settings.IsMaster { @@ -157,24 +157,42 @@ func (m *SyncManager) RunQuotaSync() int { logger.LogDebug(nil, "[RegionSync] RunQuotaSync: syncing %d users", len(users)) + // 建立 remoteUserId -> local User 映射 + userMap := make(map[int]*model.User, len(users)) + remoteIds := make([]int, len(users)) + for i := range users { + userMap[users[i].RemoteUserId] = &users[i] + remoteIds[i] = users[i].RemoteUserId + } + syncedCount := 0 - for _, user := range users { - resp, err := m.client.QueryQuota(user.RemoteUserId) - if err != nil { - common.SysError(fmt.Sprintf("[SyncManager] QueryQuota failed for user %d: %v", user.Id, err)) - continue + batchSize := 500 + for i := 0; i < len(remoteIds); i += batchSize { + end := i + batchSize + if end > len(remoteIds) { + end = len(remoteIds) } - if !resp.Success { + batch := remoteIds[i:end] + + resp, err := m.client.BatchQueryQuota(batch) + if err != nil { + common.SysError(fmt.Sprintf("[SyncManager] BatchQueryQuota failed (batch %d-%d): %v", i, end, err)) continue } - oldQuota := user.SyncedQuota - if err := model.UpdateSyncedQuota(user.Id, resp.Quota); err != nil { - common.SysError(fmt.Sprintf("[SyncManager] UpdateSyncedQuota failed for user %d: %v", user.Id, err)) - continue + + for _, entry := range resp.Quotas { + if user, ok := userMap[entry.UserId]; ok { + oldQuota := user.SyncedQuota + if err := model.UpdateSyncedQuota(user.Id, entry.Quota); err != nil { + common.SysError(fmt.Sprintf("[SyncManager] UpdateSyncedQuota failed for user %d: %v", user.Id, err)) + continue + } + logger.LogDebug(nil, "[RegionSync] RunQuotaSync: userId=%d, syncedQuota %d -> %d", user.Id, oldQuota, entry.Quota) + syncedCount++ + } } - logger.LogDebug(nil, "[RegionSync] RunQuotaSync: userId=%d, syncedQuota %d -> %d", user.Id, oldQuota, resp.Quota) - syncedCount++ } + logger.LogDebug(nil, "[RegionSync] RunQuotaSync: completed, synced=%d/%d", syncedCount, len(users)) return syncedCount } diff --git a/service/region_sync/sync_types.go b/service/region_sync/sync_types.go index d60a13b..31e18ed 100644 --- a/service/region_sync/sync_types.go +++ b/service/region_sync/sync_types.go @@ -81,3 +81,21 @@ type SyncConfigResponse struct { SyncIntervalSeconds int `json:"sync_interval_seconds"` Error string `json:"error,omitempty"` } + +// QuotaEntry 单个用户的余额信息 +type QuotaEntry struct { + UserId int `json:"user_id"` + Quota int `json:"quota"` +} + +// BatchQueryQuotaRequest 批量查询余额请求 +type BatchQueryQuotaRequest struct { + UserIds []int `json:"user_ids"` +} + +// BatchQueryQuotaResponse 批量查询余额响应 +type BatchQueryQuotaResponse struct { + Success bool `json:"success"` + Quotas []QuotaEntry `json:"quotas"` + Error string `json:"error,omitempty"` +} diff --git a/update-image.sh b/update-image.sh new file mode 100644 index 0000000..cca68f4 --- /dev/null +++ b/update-image.sh @@ -0,0 +1,59 @@ +#!/bin/bash +# 用法: ./update-image.sh [目录1] [目录2] ... +# 示例: ./update-image.sh /opt/ov /opt/cn 202604071354-feat-region-sync-glm +# +# 脚本会修改指定目录下 docker-compose.yml 中 new-api 服务的 image tag。 +# 如果不传目录参数,默认修改当前目录。 + +set -e + +if [ $# -lt 1 ]; then + echo "错误: 请提供新的 tag" + echo "用法: $0 [目录1] [目录2] ... " + echo "示例: $0 /opt/ov /opt/cn 202604071354-feat-region-sync-glm" + exit 1 +fi + +TAG="${@: -1}" + +if [ $# -gt 1 ]; then + DIRS=("${@:1:$#-1}") +else + DIRS=(".") +fi + +update_file() { + local FILE="$1" + local TAG="$2" + + if [ ! -f "$FILE" ]; then + echo "跳过: $FILE 不存在" + return + fi + + OLD_IMAGE=$(grep -E '^\s+image:.*new-api' "$FILE" | head -1 | sed 's/.*image: *//') + OLD_TAG="${OLD_IMAGE##*:}" + REPO="${OLD_IMAGE%:*}" + + if [ -z "$OLD_TAG" ] || [ -z "$REPO" ]; then + echo "错误: 无法从 $FILE 中解析 new-api 的 image 配置" + return 1 + fi + + NEW_IMAGE="${REPO}:${TAG}" + + if [[ "$OSTYPE" == "darwin"* ]]; then + perl -i -pe "s|(\s+image:) *${OLD_IMAGE}|\$1 ${NEW_IMAGE}|" "$FILE" + else + sed -i "s|${OLD_IMAGE}|${NEW_IMAGE}|" "$FILE" + fi + + echo "已更新 $FILE:" + echo " 旧: ${OLD_IMAGE}" + echo " 新: ${NEW_IMAGE}" +} + +for DIR in "${DIRS[@]}"; do + FILE="${DIR%/}/docker-compose.yml" + update_file "$FILE" "$TAG" +done