89 lines
3.1 KiB
Go
89 lines
3.1 KiB
Go
package user
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
userdomain "hyapp/services/user-service/internal/domain/user"
|
||
)
|
||
|
||
const (
|
||
defaultRegionRebuildPollInterval = 5 * time.Second
|
||
defaultRegionRebuildLockTTL = 30 * time.Second
|
||
defaultRegionRebuildBatchSize = 500
|
||
)
|
||
|
||
// RegionRebuildWorkerOptions 控制 user_region_rebuild_tasks 的后台消费节奏。
|
||
type RegionRebuildWorkerOptions struct {
|
||
// WorkerID 写入 task locked_by，便于排查某个节点是否卡住任务。
|
||
WorkerID string
|
||
// PollInterval 是没有任务或单轮处理后下一次扫描的间隔。
|
||
PollInterval time.Duration
|
||
// LockTTL 是单个 task claim 后允许本节点持有的最长时间，过期可被其他节点接管。
|
||
LockTTL time.Duration
|
||
// BatchSize 是单次最多更新的 users 数量，避免大国家全量更新长事务。
|
||
BatchSize int
|
||
}
|
||
|
||
func normalizeRegionRebuildWorkerOptions(options RegionRebuildWorkerOptions) RegionRebuildWorkerOptions {
|
||
options.WorkerID = strings.TrimSpace(options.WorkerID)
|
||
if options.WorkerID == "" {
|
||
options.WorkerID = "user-region-rebuild"
|
||
}
|
||
if options.PollInterval <= 0 {
|
||
options.PollInterval = defaultRegionRebuildPollInterval
|
||
}
|
||
if options.LockTTL <= 0 {
|
||
options.LockTTL = defaultRegionRebuildLockTTL
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
options.BatchSize = defaultRegionRebuildBatchSize
|
||
}
|
||
|
||
return options
|
||
}
|
||
|
||
// RunRegionRebuildWorker 持续消费 user_region_rebuild_tasks，保证管理端映射变更最终刷新历史用户快照。
|
||
func (s *Service) RunRegionRebuildWorker(ctx context.Context, options RegionRebuildWorkerOptions) {
|
||
options = normalizeRegionRebuildWorkerOptions(options)
|
||
if s.regionRebuildRepository == nil {
|
||
log.Printf("worker=user_region_rebuild status=stopped reason=%q", "repository is not configured")
|
||
return
|
||
}
|
||
|
||
_, _ = s.ProcessNextRegionRebuildTask(ctx, options)
|
||
ticker := time.NewTicker(options.PollInterval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
_, _ = s.ProcessNextRegionRebuildTask(ctx, options)
|
||
}
|
||
}
|
||
}
|
||
|
||
// ProcessNextRegionRebuildTask 处理一个待重算任务；错误只影响本轮，worker 下一轮继续扫描。
|
||
func (s *Service) ProcessNextRegionRebuildTask(ctx context.Context, options RegionRebuildWorkerOptions) (userdomain.RegionRebuildProcessResult, error) {
|
||
options = normalizeRegionRebuildWorkerOptions(options)
|
||
if s.regionRebuildRepository == nil {
|
||
return userdomain.RegionRebuildProcessResult{NoTask: true}, nil
|
||
}
|
||
|
||
result, err := s.regionRebuildRepository.ProcessNextRegionRebuildTask(ctx, options.WorkerID, s.now().UnixMilli(), options.LockTTL, options.BatchSize)
|
||
if err != nil {
|
||
log.Printf("worker=user_region_rebuild worker_id=%s status=failed error=%q", options.WorkerID, err.Error())
|
||
return result, err
|
||
}
|
||
if result.NoTask {
|
||
return result, nil
|
||
}
|
||
log.Printf("worker=user_region_rebuild worker_id=%s task_id=%d country=%s target_region_id=%d status=%s processed_users=%d cursor_user_id=%d", options.WorkerID, result.TaskID, result.CountryCode, result.TargetRegionID, result.Status, result.ProcessedUsers, result.CursorUserID)
|
||
|
||
return result, nil
|
||
}
|