|
| 1 | +/** |
| 2 | + * Meilisearch 写入任务队列与 Worker（异步写入 + 失败重试） |
| 3 | + * 实现 redo.md 2.6：Meilisearch 写入失败补偿/重试/后台任务 |
| 4 | + */ |
| 5 | +package jobs |
| 6 | + |
| 7 | +import ( |
| 8 | + "context" |
| 9 | + "encoding/json" |
| 10 | + "fmt" |
| 11 | + "sync" |
| 12 | + "time" |
| 13 | + |
| 14 | + "short-link/internal/config" |
| 15 | + "short-link/models" |
| 16 | + "short-link/utils" |
| 17 | + |
| 18 | + "github.com/meilisearch/meilisearch-go" |
| 19 | +) |
| 20 | + |
| 21 | +// MeiliTask Meilisearch 写入任务 |
| 22 | +type MeiliTask struct { |
| 23 | + Action string `json:"action"` // "index", "delete" |
| 24 | + Link *models.Link `json:"link,omitempty"` |
| 25 | + LinkID int64 `json:"link_id,omitempty"` |
| 26 | +} |
| 27 | + |
| 28 | +// MeiliWorker Meilisearch 写入 Worker |
| 29 | +type MeiliWorker struct { |
| 30 | + taskChan chan *MeiliTask |
| 31 | + client *meilisearch.Client |
| 32 | + index *meilisearch.Index |
| 33 | + maxRetries int |
| 34 | + retryDelay time.Duration |
| 35 | + wg sync.WaitGroup |
| 36 | + ctx context.Context |
| 37 | + cancel context.CancelFunc |
| 38 | +} |
| 39 | + |
| 40 | +// NewMeiliWorker 创建 Meilisearch Worker |
| 41 | +func NewMeiliWorker(cfg *config.Config, maxRetries int, retryDelay time.Duration) (*MeiliWorker, error) { |
| 42 | + client := meilisearch.NewClient(meilisearch.ClientConfig{ |
| 43 | + Host: cfg.MeiliHost, |
| 44 | + APIKey: cfg.MeiliKey, |
| 45 | + }) |
| 46 | + |
| 47 | + // 测试连接 |
| 48 | + if _, err := client.Health(); err != nil { |
| 49 | + return nil, fmt.Errorf("Meilisearch连接失败: %w", err) |
| 50 | + } |
| 51 | + |
| 52 | + index := client.Index("links") |
| 53 | + ctx, cancel := context.WithCancel(context.Background()) |
| 54 | + |
| 55 | + return &MeiliWorker{ |
| 56 | + taskChan: make(chan *MeiliTask, 1000), // 缓冲1000个任务 |
| 57 | + client: client, |
| 58 | + index: index, |
| 59 | + maxRetries: maxRetries, |
| 60 | + retryDelay: retryDelay, |
| 61 | + ctx: ctx, |
| 62 | + cancel: cancel, |
| 63 | + }, nil |
| 64 | +} |
| 65 | + |
| 66 | +// Submit 提交 Meilisearch 写入任务（非阻塞） |
| 67 | +func (w *MeiliWorker) Submit(action string, link *models.Link, linkID int64) { |
| 68 | + task := &MeiliTask{ |
| 69 | + Action: action, |
| 70 | + Link: link, |
| 71 | + LinkID: linkID, |
| 72 | + } |
| 73 | + |
| 74 | + select { |
| 75 | + case w.taskChan <- task: |
| 76 | + // 成功提交 |
| 77 | + default: |
| 78 | + // 队列满，静默丢弃（避免阻塞主流程） |
| 79 | + utils.LogWarn("Meilisearch任务队列已满，丢弃任务: action=%s, link_id=%d", action, linkID) |
| 80 | + } |
| 81 | +} |
| 82 | + |
| 83 | +// Start 启动 Worker（后台 goroutine） |
| 84 | +func (w *MeiliWorker) Start() { |
| 85 | + w.wg.Add(1) |
| 86 | + go w.run() |
| 87 | + utils.LogInfo("Meilisearch Worker 已启动（最大重试次数=%d，重试间隔=%v）", w.maxRetries, w.retryDelay) |
| 88 | +} |
| 89 | + |
| 90 | +// run Worker 主循环 |
| 91 | +func (w *MeiliWorker) run() { |
| 92 | + defer w.wg.Done() |
| 93 | + |
| 94 | + for { |
| 95 | + select { |
| 96 | + case <-w.ctx.Done(): |
| 97 | + return |
| 98 | + |
| 99 | + case task := <-w.taskChan: |
| 100 | + w.processTask(task) |
| 101 | + } |
| 102 | + } |
| 103 | +} |
| 104 | + |
| 105 | +// processTask 处理单个任务（带重试） |
| 106 | +func (w *MeiliWorker) processTask(task *MeiliTask) { |
| 107 | + var err error |
| 108 | + for attempt := 0; attempt < w.maxRetries; attempt++ { |
| 109 | + if attempt > 0 { |
| 110 | + // 重试前等待 |
| 111 | + select { |
| 112 | + case <-w.ctx.Done(): |
| 113 | + return |
| 114 | + case <-time.After(w.retryDelay * time.Duration(attempt)): |
| 115 | + } |
| 116 | + } |
| 117 | + |
| 118 | + switch task.Action { |
| 119 | + case "index": |
| 120 | + err = w.indexLink(task.Link) |
| 121 | + case "delete": |
| 122 | + err = w.deleteLink(task.LinkID) |
| 123 | + default: |
| 124 | + utils.LogError("未知的 Meilisearch 任务类型: %s", task.Action) |
| 125 | + return |
| 126 | + } |
| 127 | + |
| 128 | + if err == nil { |
| 129 | + // 成功 |
| 130 | + return |
| 131 | + } |
| 132 | + |
| 133 | + utils.LogWarn("Meilisearch写入失败（尝试 %d/%d）: action=%s, link_id=%d, error=%v", |
| 134 | + attempt+1, w.maxRetries, task.Action, task.LinkID, err) |
| 135 | + } |
| 136 | + |
| 137 | + // 所有重试都失败，记录错误（后续可扩展为死信队列） |
| 138 | + utils.LogError("Meilisearch写入最终失败: action=%s, link_id=%d, error=%v", task.Action, task.LinkID, err) |
| 139 | +} |
| 140 | + |
| 141 | +// indexLink 索引链接到 Meilisearch |
| 142 | +func (w *MeiliWorker) indexLink(link *models.Link) error { |
| 143 | + if link == nil { |
| 144 | + return fmt.Errorf("link 不能为空") |
| 145 | + } |
| 146 | + |
| 147 | + doc := map[string]interface{}{ |
| 148 | + "id": link.ID, |
| 149 | + "code": link.Code, |
| 150 | + "original_url": link.OriginalURL, |
| 151 | + "title": link.Title, |
| 152 | + "user_id": link.UserID, |
| 153 | + "domain_id": link.DomainID, |
| 154 | + "created_at": link.CreatedAt.Unix(), |
| 155 | + } |
| 156 | + |
| 157 | + _, err := w.index.AddDocuments([]map[string]interface{}{doc}, "id") |
| 158 | + return err |
| 159 | +} |
| 160 | + |
| 161 | +// deleteLink 从 Meilisearch 删除链接 |
| 162 | +func (w *MeiliWorker) deleteLink(linkID int64) error { |
| 163 | + _, err := w.index.DeleteDocument(fmt.Sprintf("%d", linkID)) |
| 164 | + return err |
| 165 | +} |
| 166 | + |
| 167 | +// Stop 停止 Worker |
| 168 | +func (w *MeiliWorker) Stop() { |
| 169 | + w.cancel() |
| 170 | + w.wg.Wait() |
| 171 | + utils.LogInfo("Meilisearch Worker 已停止") |
| 172 | +} |
| 173 | + |
| 174 | + |
0 commit comments