4. 异步任务模型：任务状态机设计、任务队列、Worker池、结果回调

在 Kubernetes 环境中编排数据库迁移，核心挑战在于管理长时间运行、状态敏感且可能失败的任务。本章将深入异步任务模型的四个关键组件：任务状态机、任务队列、Worker 池和结果回调，构建一个可靠、可扩展的迁移执行引擎。

4.1 任务状态机设计

每个迁移任务在其生命周期中经历多个状态。状态机确保状态转换的合法性和可追溯性，避免因并发或异常导致的状态混乱。

4.1.1 状态定义与转换矩阵

4.1.2 状态机实现示例（Go 语言）

type TaskState string

const (
    StatePending   TaskState = "PENDING"
    StateQueued    TaskState = "QUEUED"
    StateRunning   TaskState = "RUNNING"
    StateSucceeded TaskState = "SUCCEEDED"
    StateFailed    TaskState = "FAILED"
    StateCancelled TaskState = "CANCELLED"
    StateCancelling TaskState = "CANCELLING"
)

var validTransitions = map[TaskState][]TaskState{
    StatePending:   {StateQueued, StateCancelled},
    StateQueued:    {StateRunning, StateCancelled},
    StateRunning:   {StateSucceeded, StateFailed, StateCancelling},
    StateCancelling: {StateCancelled, StateFailed},
    StateFailed:    {StatePending}, // 重试
}

func (s TaskState) CanTransitionTo(next TaskState) bool {
    for _, valid := range validTransitions[s] {
        if valid == next {
            return true
        }
    }
    return false
}

4.2 任务队列

任务队列作为生产者和消费者之间的缓冲层，负责解耦任务提交与执行。在 Kubernetes 环境中，我们通常使用 Redis 或 RabbitMQ 作为队列后端，但也可以利用 Kubernetes 原生资源实现轻量级队列。

4.2.1 队列选型对比

4.2.2 基于 Redis 的队列实现

// 生产者：将任务推入队列
func EnqueueTask(redisClient *redis.Client, task *Task) error {
    taskData, _ := json.Marshal(task)
    // 使用 LPUSH 将任务放入队列头部
    err := redisClient.LPush(ctx, "migration:queue", taskData).Err()
    if err != nil {
        return err
    }
    // 更新任务状态为 QUEUED
    return updateTaskState(redisClient, task.ID, StateQueued)
}

// 消费者：从队列阻塞获取任务
func DequeueTask(redisClient *redis.Client) (*Task, error) {
    // 使用 BRPOP 阻塞等待，超时时间 30 秒
    result, err := redisClient.BRPop(ctx, 30*time.Second, "migration:queue").Result()
    if err != nil {
        return nil, err
    }
    var task Task
    json.Unmarshal([]byte(result[1]), &task)
    return &task, nil
}

4.3 Worker 池

Worker 池管理一组并发执行任务的 Worker 实例，控制并发度、资源消耗和任务分配。在 Kubernetes 中，Worker 池可以部署为 Deployment 或 StatefulSet，通过环境变量或 ConfigMap 配置池大小。

4.3.1 Worker 池架构

4.3.2 Worker 池配置与扩缩容

Worker 池的核心参数包括：

• 最小/最大 Worker 数：根据集群资源限制设置，避免资源争抢。
• 任务超时时间：单个任务的最大执行时间，超时后标记为 FAILED。
• 重试策略：失败任务的重试次数和间隔（如指数退避：1s, 2s, 4s, 8s...）。
• 优雅关闭：Pod 被删除时，等待当前任务完成或超时后再退出。

// Worker 池核心逻辑（Go 伪代码）
type WorkerPool struct {
    maxWorkers   int
    taskQueue    chan *Task
    resultChan   chan *TaskResult
    wg           sync.WaitGroup
    shutdown     chan struct{}
}

func (wp *WorkerPool) Start() {
    for i := 0; i < wp.maxWorkers; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for {
        select {
        case task := <-wp.taskQueue:
            result := executeTask(task) // 执行迁移
            wp.resultChan <- result     // 发送结果
        case <-wp.shutdown:
            return // 优雅退出
        }
    }
}

func (wp *WorkerPool) Shutdown() {
    close(wp.shutdown)
    wp.wg.Wait() // 等待所有 Worker 完成
}

4.4 结果回调

任务执行完成后，需要将结果通知给调用方或触发后续流程。结果回调机制支持同步和异步两种模式。

4.4.1 回调模式对比

4.4.2 回调实现示例（Webhook + 幂等性）

// 结果回调结构体
type CallbackPayload struct {
    TaskID      string            `json:"task_id"`
    State       TaskState         `json:"state"`
    StartTime   time.Time         `json:"start_time"`
    EndTime     time.Time         `json:"end_time"`
    Duration    string            `json:"duration"`
    Error       string            `json:"error,omitempty"`
    Metadata    map[string]string `json:"metadata"`
}

// 发送回调（带重试和幂等性）
func SendCallback(callbackURL string, payload *CallbackPayload) error {
    body, _ := json.Marshal(payload)
    // 使用任务 ID 作为幂等键
    req, _ := http.NewRequest("POST", callbackURL, bytes.NewBuffer(body))
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("Idempotency-Key", payload.TaskID)

    client := &http.Client{Timeout: 10 * time.Second}
    resp, err := client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    if resp.StatusCode >= 500 {
        return fmt.Errorf("server error: %d", resp.StatusCode)
    }
    return nil
}

4.5 完整异步任务生命周期

将上述组件整合，一个迁移任务的完整生命周期如下：

1. 提交：用户通过 API 提交迁移任务，任务进入 PENDING 状态。
2. 入队：调度器将任务序列化后推入 Redis 队列，状态变为 QUEUED。
3. 分发：分发器从队列中取出任务，分配给空闲 Worker。
4. 执行：Worker 领取任务，状态变为 RUNNING，开始执行迁移步骤。
5. 完成/失败：迁移完成后，Worker 将结果写入结果通道。
6. 回调：回调处理器读取结果，更新状态存储，并触发 Webhook 通知。
7. 重试（可选）：如果任务失败且未超过最大重试次数，状态回退到 PENDING，重新入队。