FastAPI + Redis 高性能任务队列实现：AI内容生成系统实践

引言

在现代应用中，任务队列是处理资源密集型操作的重要组件。本文将详细介绍一个基于FastAPI和Redis实现的高性能任务队列系统，该系统用于处理AI图片和视频的生成请求。我们将从架构设计、技术实现到性能优化策略进行全面分析，希望能为构建类似系统的开发者提供有益参考。

系统概述

该项目是一个AI图片和视频生成后端服务，用于处理用户的生成请求，调用国内顶尖AI模型（如统一万象、豆包/即梦AI）实现内容生成。系统的核心挑战在于：

处理时间不确定的AI生成任务
合理分配有限的计算资源
实现用户级别的队列限制
优化系统整体吞吐量和响应时间

任务队列架构设计

整体架构

任务队列系统由以下几个主要组件构成：

API层：接收用户请求，验证权限，提交任务到队列
队列服务：基于Redis实现的分布式任务队列
处理器服务：多Worker并行处理不同类型的任务
数据存储：任务和生成内容的持久化存储
OSS服务：生成内容的对象存储服务

整个流程如下：

用户提交生成请求
API层验证并创建任务记录
任务入队到Redis相应类型队列
专用Worker从队列获取任务并处理
处理结果上传到OSS并更新数据库
用户可查询任务状态和获取结果

Redis队列设计

该系统采用了按任务类型分离的Redis队列设计，而非传统的单一队列：

+-----------------+      +-------------------+      +---------------+
| task_queue:image| <--- | 图片处理Worker组  | ---> | 图片生成模型  |
+-----------------+      +-------------------+      +---------------++-----------------+      +-------------------+      +---------------+
| task_queue:video| <--- | 视频处理Worker组  | ---> | 视频生成模型  |
+-----------------+      +-------------------+      +---------------++-----------------+      +-------------------+
| task_queue:default <-- | 通用Worker        |
+-----------------+      +-------------------+

队列键设计

按任务类型分离的队列：
- task_queue:image - 图片生成任务队列
- task_queue:video - 视频生成任务队列
- task_queue:default - 默认队列（用于未指定类型的任务）
任务状态和结果存储：
- task_status:{task_id} - 存储任务状态
- task_result:{task_id} - 存储任务结果

所有状态和结果都设置了300秒（5分钟）的过期时间，避免无用数据占用内存。

技术实现详解

任务数据结构

任务在Redis中存储为JSON字符串，包含以下主要字段：

{"task_id": "uuid字符串","data": {"task_id": "uuid字符串","task_type": "image或video","parameters": {"prompt": "生成提示词","size": "图片尺寸","model_type": "模型类型","model_version": "模型版本","...": "其他参数"},"user_id": "用户ID"},"enqueue_time": "ISO时间字符串"
}

任务提交流程

async def submit_task(self, db: AsyncSession, user_id: int, task_type: TaskType, parameters: Dict[str, Any]) -> Dict[str, Any]:# 检查用户当前任务数限制current_task_count = await TaskDAO.count_user_active_tasks(db, user_id)if current_task_count >= self.max_concurrent_tasks:return {"status": "error", "message": f"已达到最大并发任务数限制({self.max_concurrent_tasks})"}# 创建任务记录task = await TaskDAO.create_task(db=db, user_id=user_id, task_type=task_type, parameters=parameters)# 将任务加入Redis队列task_data = {"task_id": task.id,"task_type": task_type.value,"parameters": parameters,"user_id": user_id}results = await queue_service.enqueue_task(task.id, task_data)if not results:# 如果入队失败，更新任务状态为失败await TaskDAO.update_task_status(db=db, task_id=task.id, status=TaskStatus.FAILED, error_message="任务入队失败")return {"status": "error", "message": "任务提交失败"}return {"status": "success", "task_id": task.id, "message": "任务已提交"}

队列操作实现

Redis队列的核心操作包括：

入队操作：使用LPUSH将任务添加到队列左侧
```
await redis.lpush(queue_key, json.dumps(task_item))
```
出队操作：使用RPOP从队列右侧获取任务
```
task_item = await redis.rpop(queue_key)
```

状态更新：使用SET存储任务状态，并设置过期时间

await redis.set(key, json.dumps(status_data), ex=self.task_expire_time)

优先级处理：高优先级任务使用RPUSH添加到队列右侧，优先被处理

if priority <= 0:# 高优先级，入队到队列右侧await redis.rpush(queue_key, json.dumps(task_item))
else:# 普通优先级，入队到队列左侧await redis.lpush(queue_key, json.dumps(task_item))

多Worker并行处理机制

Worker分配策略

系统根据任务特性，采用非均衡的Worker分配策略：

# 计算各类型worker数量
image_workers = max(1, int(self.worker_count * self.image_worker_ratio))
video_workers = max(1, self.worker_count - image_workers - 1)  # 减去1个通用worker# 创建专门的图片任务worker
for i in range(image_workers):worker = asyncio.create_task(self._process_queue_worker(i, "image"))self._workers.append(worker)# 创建专门的视频任务worker
for i in range(video_workers):worker = asyncio.create_task(self._process_queue_worker(i + image_workers, "video"))self._workers.append(worker)# 创建一个通用worker处理任何类型的任务
general_worker = asyncio.create_task(self._process_queue_worker(self.worker_count, None))
self._workers.append(general_worker)

默认情况下，系统会分配75%的Worker处理图片任务，剩余Worker处理视频任务，并保留一个通用Worker可处理任何类型的任务。

自适应轮询机制

为了减少系统资源消耗，我们实现了自适应轮询间隔策略：

# 自适应调整轮询间隔 - 指数退避策略
backoff_factor = min(stats["empty_queue_count"], 10)  # 限制退避因子
new_interval = self.base_poll_interval * (1.5 ** backoff_factor)
stats["current_interval"] = min(self.max_poll_interval, new_interval)

当队列为空时，轮询间隔会逐渐增加（采用指数退避算法），最高可达5秒；一旦有新任务，轮询间隔立即恢复到最小值（0.2秒）。这样既保证了系统对新任务的快速响应，又避免了频繁空轮询导致的资源浪费。

性能优化策略

1. 按任务类型分离队列

本项目摒弃了传统的单一主队列设计，采用完全分离的类型队列：

优势：不同类型任务可以并行处理，避免长时间任务（如视频生成）阻塞短时间任务（如图片生成）
实现：将任务直接入队到对应类型队列，而非先入队到主队列再分发

2. 专用Worker与资源分配

根据不同任务的特性，我们采用非均衡的资源分配：

图片生成任务通常较快（数秒到数十秒），但请求量大
视频生成任务较慢（数分钟），但请求量相对较小

因此，系统为图片任务分配更多Worker（默认75%），提高整体吞吐量。

3. Worker跨队列处理

当专用Worker空闲时，允许它们处理其他类型的任务：

# 如果指定类型队列为空且worker有指定类型，尝试从其他队列获取任务
if task_type and self.is_running:for other_type in self.task_types:if other_type != task_type:general_task = await queue_service.dequeue_task(other_type)if general_task:# 处理其他队列任务# ...

这种设计提高了Worker利用率，尤其在负载不均衡时效果显著。

4. 自适应轮询与指数退避

该项目使用指数退避算法动态调整轮询间隔，显著降低了系统资源消耗：

队列为空时，轮询间隔从1秒开始，逐渐增加到最大5秒
处理任务后，立即恢复到最小间隔0.2秒，保证响应速度
限制最大退避因子为10，防止间隔增长过大

5. 细粒度性能监控

系统每分钟记录详细的Worker性能指标：

uptime = current_time - stats["start_time"]
tasks_per_minute = stats["tasks_processed"] / (uptime / 60) if uptime > 0 else 0
logger.info(f"{worker_name} 指标: 处理任务数={stats['tasks_processed']}, "f"当前轮询间隔={stats['current_interval']:.2f}秒, "f"任务处理速率={tasks_per_minute:.2f}任务/分钟")

这些指标有助于发现性能瓶颈并进行针对性优化。

系统状态监控

为了便于运维和调试，系统提供了全面的状态信息API：

async def get_status(self) -> Dict[str, Any]:queue_stats = await queue_service.get_queue_stats()# 计算worker类型分布worker_types = {"image": 0, "video": 0, "default": 0}# ...# 收集worker性能指标worker_performance = {}# ...return {"is_running": self.is_running,"worker_count": len(self._worker_stats),"worker_types": worker_types,"queue_stats": queue_stats,"performance": {"total_tasks_processed": sum(stats["tasks_processed"] for stats in self._worker_stats.values()),"avg_poll_interval": round(sum(stats["current_interval"] for stats in self._worker_stats.values()) / max(1, len(self._worker_stats)), 2),"worker_stats": worker_performance}}