4. 任务调度与向量检索结合：向量检索任务的生命周期
大家好，我是老张。今天我们来聊聊向量检索任务调度里最核心的一块——任务的生命周期管理。说白了，就是一条查询请求从进门到出门，中间经历了什么，每一步该怎么管。
我在做第一个分布式向量检索系统时，踩过一个大坑。当时任务跑着跑着就卡死了，查了半天才发现是生命周期状态机没设计好，死锁了。嗯，从那以后我对这块就特别较真。
4.1 向量检索任务的生命周期
一个典型的向量检索任务，会经历这么几个阶段：
- 创建（Created）：客户端提交查询请求，系统生成任务ID。
- 排队（Queued）：任务进入调度队列，等待资源。
- 调度（Scheduled）：调度器为任务分配计算节点。
- 执行（Running）：任务在节点上实际执行向量检索。
- 完成（Completed）：返回结果给客户端。
- 失败（Failed）：执行过程中出现异常。
- 超时（TimedOut）：超过预设时间仍未完成。
关键点：状态转换必须是单向的，不能出现循环。比如从Running回到Queued，这种设计会引发死锁。我见过有人这么干，结果集群直接瘫痪。
下面这张图，是我个人习惯用的任务状态机。你看一眼就明白了：
我的经验：状态机里一定要加一个「取消（Cancelled）」状态。用户等不及了想取消任务，你得有地方放。我曾经漏了这个，结果用户投诉说任务取消不了，只能等它超时。
4.2 任务拆分与合并
向量检索有个特点：数据量大了，单机扛不住。所以得把一个大任务拆成多个小任务，分给不同节点去跑。跑完了再把结果合并回来。
你想想看，一个1000万条向量的数据集，你不可能让一台机器全量扫描。拆！必须拆！
4.2.1 拆分策略
我个人习惯用这几种拆分方式：
| 拆分方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 按分片拆分 | 数据已分片存储 | 天然隔离，无数据重叠 | 分片不均匀时负载倾斜 |
| 按向量ID范围拆分 | 连续ID的数据集 | 实现简单 | ID分布不均时效果差 |
| 按哈希值拆分 | 任意数据集 | 分布均匀 | 需要额外计算哈希 |
| 按聚类中心拆分 | IVF索引 | 减少候选集大小 | 聚类质量影响大 |
注意：拆分粒度不能太细。我见过有人把任务拆成10000个子任务，结果调度开销比执行时间还长。一般来说，子任务执行时间控制在100ms-1s之间比较合理。
4.2.2 合并策略
子任务跑完了，怎么合并？这里有个坑：
- Top-K合并：每个子任务返回自己的Top-K结果，然后全局排序取前K个。这是最常用的方式。
- 阈值合并：每个子任务返回距离小于阈值的结果，然后去重合并。适合召回率要求高的场景。
- 加权合并：不同分片的数据重要性不同，给结果加权后再合并。我曾在推荐系统里用过这个。
代码实现其实不复杂，核心就是一个多路归并：
// 多路归并合并Top-K结果
func mergeTopK(results [][]Result, k int) []Result {
h := &minHeap{}
for i, list := range results {
if len(list) > 0 {
heap.Push(h, &item{result: list[0], listIdx: i, elemIdx: 0})
}
}
merged := make([]Result, 0, k)
for h.Len() > 0 && len(merged) < k {
it := heap.Pop(h).(*item)
merged = append(merged, it.result)
nextIdx := it.elemIdx + 1
if nextIdx < len(results[it.listIdx]) {
heap.Push(h, &item{
result: results[it.listIdx][nextIdx],
listIdx: it.listIdx,
elemIdx: nextIdx,
})
}
}
return merged
}避坑指南：合并时要注意去重。同一个向量可能出现在多个分片里，尤其是用IVF索引时。我曾经没做去重，结果返回的结果里有一半是重复的，被客户骂惨了。
4.3 任务优先级管理
不是所有任务都平等的。线上实时查询和离线批量分析，优先级能一样吗？
我建议把任务分成三个优先级：
- 高优先级：在线实时查询，延迟要求<10ms。这类任务要抢占资源。
- 中优先级：准实时查询，延迟要求<100ms。可以排队，但不能等太久。
- 低优先级：离线批量任务，没有严格延迟要求。用空闲资源跑就行。
实现优先级调度，我推荐用多级队列：
type PriorityScheduler struct {
highQueue chan *Task // 高优先级，容量1000
mediumQueue chan *Task // 中优先级，容量5000
lowQueue chan *Task // 低优先级，容量10000
}
func (s *PriorityScheduler) Schedule() *Task {
select {
case task := <-s.highQueue:
return task
case task := <-s.mediumQueue:
return task
case task := <-s.lowQueue:
return task
default:
return nil
}
}注意：要防止低优先级任务饿死。我习惯的做法是：如果低优先级任务排队超过30秒，就提升它的优先级。这叫「老化机制」。
4.4 任务超时与重试机制
分布式系统里，失败是常态。网络抖动、节点宕机、索引加载慢……什么情况都可能发生。
4.4.1 超时设置
超时时间怎么设？我总结了一个公式：
超时时间 = 预估执行时间 × 系数
系数一般取1.5到3之间。线上查询取1.5，离线任务取3。为什么？线上查询等不起，超时就赶紧返回降级结果；离线任务可以多等一会儿，避免频繁重试。
| 任务类型 | 预估执行时间 | 超时系数 | 实际超时 |
|---|---|---|---|
| 在线Top-10查询 | 5ms | 1.5 | 7.5ms |
| 批量聚类 | 30s | 3 | 90s |
| 索引构建 | 10min | 2 | 20min |
4.4.2 重试策略
重试不是简单的「失败了就再来一次」。我见过有人这么干，结果系统雪崩了。
正确的做法是：
- 指数退避：第一次重试等100ms，第二次等200ms，第三次等400ms……
- 最大重试次数：一般设3-5次。超过次数就标记为失败。
- 重试范围控制：只重试幂等的子任务。如果子任务已经写入了部分结果，重试会导致数据不一致。
核心原则：重试只适用于临时性故障。如果是永久性故障（比如节点宕机），重试多少次都没用，应该直接走降级策略。
下面是一个重试机制的伪代码，我在项目里就是这么用的：
func retryWithBackoff(task *Task, maxRetries int) error {
for i := 0; i < maxRetries; i++ {
err := executeTask(task)
if err == nil {
return nil
}
// 判断是否值得重试
if !isRetryable(err) {
return err
}
// 指数退避：100ms, 200ms, 400ms, 800ms, 1600ms
wait := time.Duration(100 << uint(i)) * time.Millisecond
time.Sleep(wait)
log.Printf("任务 %s 第 %d 次重试，等待 %v", task.ID, i+1, wait)
}
return fmt.Errorf("任务 %s 重试 %d 次后仍然失败", task.ID, maxRetries)
}
func isRetryable(err error) bool {
// 网络超时、节点繁忙 -> 可重试
// 数据不存在、权限不足 -> 不可重试
switch err.(type) {
case *NetworkTimeoutError, *NodeBusyError:
return true
default:
return false
}
}我的经验：重试时要加「去重」机制。同一个任务可能被多个调度器同时重试，导致重复执行。我习惯用Redis的分布式锁来保证只有一个副本在执行。
4.5 小结
好了，这一章的内容就这些。总结一下：
- 任务生命周期要设计成单向状态机，别搞循环。
- 拆分和合并要平衡粒度，太细了调度开销大，太粗了负载不均。
- 优先级管理用多级队列，别忘了老化机制。
- 超时和重试要配合使用，指数退避是标配。
这些经验都是我在实际项目中一点点踩坑踩出来的。你照着做，至少能避免80%的调度问题。剩下的20%，等你遇到了再来找我聊。