🧠 图解记忆：任务状态要持久化，消费要幂等，重试有上限，失败进入死信并可恢复；点击图片可查看原图。
💡 答案要点
题目理解：
AI 任务队列：
- 异步处理：AI 生成耗时长，不能同步等
- 避免超时：长任务不超时、不丢失
- 保证顺序：同一用户的请求要按顺序处理
- 核心挑战：可靠性 + 延迟 + 顺序保证问题分析：
AI 任务的特点：
1. 耗时长（10s-5min）→ 不能同步等 → 必须异步
2. LLM 输出不稳定 → 可能中途失败 → 需要重试
3. 同一用户多请求 → 不能乱序 → 需要顺序保证
4. 结果要推送 → 不能轮询 → 需要 WebSocket/SSE整体架构：
客户端
↓ 发起请求
┌─────────────────────────────┐
│ API 服务 │
│ 1. 快速返回 task_id │
│ 2. 写入任务队列 │
│ 3. 建立 WebSocket 连接 │
└─────────────────────────────┘
↓ 写入
┌─────────────────────────────┐
│ Redis / RabbitMQ │
│ 任务队列 + 顺序保证 │
└─────────────────────────────┘
↓ 消费
┌─────────────────────────────┐
│ Worker 集群 │
│ 调用 LLM │ 重试 │ 超时处理 │
└─────────────────────────────┘
↓ 写入结果
┌─────────────────────────────┐
│ Redis / 数据库 │
│ 任务结果存储 │
└─────────────────────────────┘
↓ 推送
WebSocket / SSE → 客户端实现方案 1：Redis 延迟队列（简单场景）：
展开 Python 代码示例（85 行）
python
import redis
import json
import time
import asyncio
from dataclasses import dataclass
from typing import Optional
@dataclass
class AITask:
task_id: str
user_id: str
prompt: str
priority: int = 0
created_at: float = None
def __post_init__(self):
if self.created_at is None:
self.created_at = time.time()
class AITaskQueue:
"""基于 Redis 的 AI 任务队列"""
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url)
self.queue_key = "ai:tasks:pending"
self.processing_key = "ai:tasks:processing"
self.result_prefix = "ai:tasks:result:"
self.user_order_prefix = "ai:tasks:user:"
def enqueue(self, task: AITask) -> str:
"""入队，返回 task_id"""
# 优先级队列：分数 = -priority, timestamp
score = task.created_at - task.priority * 10000000
self.redis.zadd(self.queue_key, {json.dumps(task.__dict__): score})
# 记录用户的任务顺序
self.redis.zadd(f"{self.user_order_prefix}{task.user_id}", {task.task_id: task.created_at})
return task.task_id
def dequeue(self, timeout: int = 5) -> Optional[AITask]:
"""出队（阻塞），保证同一用户顺序"""
# 先尝试获取任务
result = self.redis.zpopmin(self.queue_key, 1)
if not result:
# 用 BZPOPMIN 阻塞等待
result = self.redis.bzpopmin(self.queue_key, timeout)
if not result:
return None
task_data = json.loads(result[0])
task = AITask(**task_data)
# 加入处理中队列（用于超时检测）
self.redis.zadd(self.processing_key, {task.task_id: time.time()})
return task
def mark_done(self, task_id: str, result: dict):
"""标记任务完成"""
# 从处理中移除
self.redis.zrem(self.processing_key, task_id)
# 写入结果
self.redis.setex(
f"{self.result_prefix}{task_id}",
3600, # 1 小时过期
json.dumps(result)
)
def mark_failed(self, task_id: str, error: str):
"""标记任务失败，触发重试"""
self.redis.zrem(self.processing_key, task_id)
# 重试队列
self.redis.zadd(self.queue_key, {
json.dumps({"task_id": task_id, "error": error}): time.time()
})
def get_result(self, task_id: str) -> Optional[dict]:
"""获取任务结果"""
return self.redis.get(f"{self.result_prefix}{task_id}")实现方案 2：Celery（生产级）：
展开 Python 代码示例（44 行）
python
# tasks.py
from celery import Celery
from celery.signals import task_prerun, task_postrun, task_failure
app = Celery('ai_tasks', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def generate_text_task(self, prompt: str, user_id: str):
"""
AI 文本生成任务
- max_retries=3：最多重试 3 次
- default_retry_delay=60：重试间隔 60 秒
"""
try:
# 调用 LLM
response = llm_client.chat(prompt)
return {"status": "success", "result": response}
except RateLimitError as exc:
# 限流错误：60 秒后重试
raise self.retry(exc=exc, countdown=60)
except TimeoutError as exc:
# 超时错误：30 秒后重试
raise self.retry(exc=exc, countdown=30)
except Exception as exc:
# 其他错误：不重试
return {"status": "error", "error": str(exc)}
@app.task
def batch_generate_task(prompts: list[str], user_id: str) -> list[dict]:
"""批量生成任务（保证顺序）"""
results = []
for i, prompt in enumerate(prompts):
# 顺序执行（不并发）
result = generate_text_task.apply(
args=[prompt, user_id],
task_id=f"{user_id}:batch:{i}" # 指定 task_id 保证顺序
)
results.append(result)
return results超时保证机制：
展开 Python 代码示例（37 行）
python
class TimeoutHandler:
"""任务超时处理"""
def __init__(self, queue: AITaskQueue, timeout: int = 300):
self.queue = queue
self.timeout = timeout
self.check_interval = 30 # 每 30 秒检查
async def start(self):
"""启动超时检测"""
while True:
await self.check_timeouts()
await asyncio.sleep(self.check_interval)
async def check_timeouts(self):
"""检查处理超时的任务"""
processing = self.queue.redis.zrange(
"ai:tasks:processing", 0, -1, withscores=True
)
now = time.time()
for task_id, start_time in processing:
if now - start_time > self.timeout:
print(f"Task {task_id} timeout, moving to retry queue")
# 标记超时
self.queue.mark_failed(task_id, "timeout")
# 增加超时计数
self.queue.redis.hincrby("ai:tasks:timeout_count", task_id, 1)
# 超时超过 3 次，放弃
timeout_count = self.queue.redis.hget(
"ai:tasks:timeout_count", task_id
)
if int(timeout_count or 0) >= 3:
await self.notify_user(task_id, "Task failed after 3 retries")保证顺序的方案：
| 方案 | 实现 | 优点 | 缺点 |
|---|---|---|---|
| 单队列 + 单 Worker | 一个 Worker 串行处理 | 简单、保证顺序 | 吞吐低 |
| 用户分区 | 同一用户路由到同一 Worker | 吞吐 + 顺序 | 负载不均 |
| 版本号乐观锁 | 每个任务带版本号，丢弃旧版本 | 并发高 | 实现复杂 |
| 优先级队列 | 同一用户任务优先级相同 | 灵活 | 需要额外协调 |
面试话术：
"AI 任务队列的核心是'异步 + 顺序 + 超时'。我用的方案是 Redis ZSet 做优先级队列 + 用户分区保证顺序。异步靠任务入队立即返回 task_id，客户端用 WebSocket 订阅结果；顺序靠任务带上版本号，消费时检查版本，过期任务直接丢弃；超时靠独立进程每 30 秒扫描处理中超时任务，超过 3 次超时自动放弃并通知用户。Celery 是生产级首选，它自带重试、死信队列、监控，比手写 Redis 队列稳定。"