RunQuotaSync 从逐个查询改为批量查询(每批500个), 减少 Master 节点 API 调用次数,提升同步效率。 新增: - BatchQueryUserQuota 控制器接口 - BatchQueryQuota 客户端方法 - QuotaEntry/BatchQueryQuotaRequest/Response 类型 附带 update-image.sh 部署脚本。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>feat/alipay-payment
| @@ -145,6 +145,41 @@ func QueryUserQuota(c *gin.Context) { | |||||
| c.JSON(http.StatusOK, gin.H{"success": true, "quota": user.Quota}) | 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 节点) | // BatchDeductQuota 批量扣费(供 slave 节点调用 master 节点) | ||||
| func BatchDeductQuota(c *gin.Context) { | func BatchDeductQuota(c *gin.Context) { | ||||
| var req region_sync.BatchDeductRequest | var req region_sync.BatchDeductRequest | ||||
| @@ -142,3 +142,15 @@ func (c *SyncClient) FetchConfig() (*SyncConfigResponse, error) { | |||||
| } | } | ||||
| return &resp, nil | 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 | |||||
| } | |||||
| @@ -143,7 +143,7 @@ func (m *SyncManager) QueryMasterQuota(remoteUserId int) (int, error) { | |||||
| return resp.Quota, nil | return resp.Quota, nil | ||||
| } | } | ||||
| // RunQuotaSync 从 Master 拉取所有同步用户的最新余额,更新本地 synced_quota | |||||
| // RunQuotaSync 从 Master 批量拉取所有同步用户的最新余额,更新本地 synced_quota | |||||
| func (m *SyncManager) RunQuotaSync() int { | func (m *SyncManager) RunQuotaSync() int { | ||||
| settings := system_setting.GetRegionSyncSettings() | settings := system_setting.GetRegionSyncSettings() | ||||
| if !settings.Enabled || settings.IsMaster { | if !settings.Enabled || settings.IsMaster { | ||||
| @@ -157,24 +157,42 @@ func (m *SyncManager) RunQuotaSync() int { | |||||
| logger.LogDebug(nil, "[RegionSync] RunQuotaSync: syncing %d users", len(users)) | 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 | 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 | 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)) | logger.LogDebug(nil, "[RegionSync] RunQuotaSync: completed, synced=%d/%d", syncedCount, len(users)) | ||||
| return syncedCount | return syncedCount | ||||
| } | } | ||||
| @@ -81,3 +81,21 @@ type SyncConfigResponse struct { | |||||
| SyncIntervalSeconds int `json:"sync_interval_seconds"` | SyncIntervalSeconds int `json:"sync_interval_seconds"` | ||||
| Error string `json:"error,omitempty"` | 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"` | |||||
| } | |||||
| @@ -0,0 +1,59 @@ | |||||
| #!/bin/bash | |||||
| # 用法: ./update-image.sh [目录1] [目录2] ... <new-tag> | |||||
| # 示例: ./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] ... <new-tag>" | |||||
| 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 | |||||