4. 异步导出任务模型：任务队列、生产者-消费者模式、任务状态机设计

在缓存失效导致数据源压力陡增的场景下，异步导出任务必须从“即时执行”转变为“排队执行”。本章围绕任务队列、生产者-消费者模式与任务状态机三个核心要素，构建一个可降级、可补偿、可追踪的异步导出任务模型。

4.1 任务队列：异步导出的缓冲核心

任务队列是异步导出系统的数据缓冲层，负责接收生产者提交的导出请求，并按顺序分发给消费者处理。在 Redis 缓存失效的降级场景中，任务队列需要具备以下能力：

• 持久化：队列中的任务不因 Redis 宕机而丢失，需支持后端存储（如 MySQL、Kafka）作为兜底。
• 优先级：支持按用户等级或任务紧急程度设置优先级，高优任务可插队。
• 去重：同一用户对同一数据集的重复导出请求，在队列层面合并为一次。
• 限流：控制队列积压深度，防止消费者被突发流量冲垮。

4.2 生产者-消费者模式：解耦与流量削峰

生产者-消费者模式将导出任务的提交与执行完全解耦。生产者只负责将任务描述（元数据）写入队列，消费者从队列中拉取任务并执行实际的导出逻辑。这种模式天然支持流量削峰：即使瞬间涌入大量导出请求，队列也能平滑消化。

4.3 任务状态机：全生命周期可追踪

每个导出任务从创建到完成（或失败），经历一系列确定的状态转换。状态机设计是任务可观测、可补偿的基础。以下为异步导出任务的标准状态机：

4.4 状态转换与补偿逻辑

状态机中的每个转换都对应明确的业务动作。在缓存失效的降级场景下，以下转换路径需要特别关注：

4.5 任务元数据设计

每个任务在队列中携带的元数据决定了消费者如何执行导出。以下为推荐的任务元数据结构：

{
  "taskId": "exp_20250321_001",
  "userId": "u_12345",
  "type": "order_export",
  "params": {
    "startDate": "2025-03-01",
    "endDate": "2025-03-21",
    "format": "xlsx"
  },
  "priority": 5,
  "status": "QUEUED",
  "retryCount": 0,
  "maxRetries": 3,
  "ttl": 300,
  "createTime": 1711000000,
  "source": "web"
}

• taskId：全局唯一，用于去重和追踪。
• type：导出类型，消费者根据 type 路由到不同的导出处理器。
• params：导出参数，包含查询条件、文件格式等。
• priority：优先级数值，数值越小优先级越高（1 最高，10 最低）。
• retryCount / maxRetries：控制重试次数，防止死循环。
• ttl：任务超时时间（秒），超时后自动标记为 CANCELLED。
• source：任务来源，用于区分正常请求与降级补偿产生的任务。

4.6 任务队列的 Redis 实现示例

以下代码展示了使用 Redis Stream 实现任务队列的生产者与消费者核心逻辑：

// 生产者：提交导出任务
public function submitExportTask(array $taskData): string
{
    $taskId = $this->generateTaskId();
    $taskData['taskId'] = $taskId;
    $taskData['status'] = 'PENDING';
    $taskData['createTime'] = time();

    // 写入 Redis Stream
    $this->redis->xAdd('export:task:stream', '*', $taskData);

    // 同时写入 MySQL 任务表做持久化兜底
    $this->mysql->insert('export_tasks', $taskData);

    return $taskId;
}

// 消费者：拉取并处理任务
public function consumeExportTask(): void
{
    // 从 Stream 中读取未确认的任务
    $messages = $this->redis->xReadGroup(
        'export:consumer:group',
        'worker_1',
        ['export:task:stream' => '>'],
        1,  // 每次只取 1 个任务
        0   // 阻塞等待
    );

    foreach ($messages as $messageId => $taskData) {
        try {
            // 更新状态为 PROCESSING
            $this->updateTaskStatus($taskData['taskId'], 'PROCESSING');

            // 执行导出逻辑
            $this->executeExport($taskData);

            // 确认消息已处理
            $this->redis->xAck('export:task:stream', 'export:consumer:group', [$messageId]);
            $this->updateTaskStatus($taskData['taskId'], 'COMPLETED');
        } catch (\Exception $e) {
            // 失败处理：重试或标记失败
            $this->handleFailure($taskData, $e);
        }
    }
}