简介
本地原型单机运行，依赖本地数据库和队列，无法应对生产环境的并发、可靠性和可观测性要求。本文基于阿里云ECS、Redis、RDS MySQL和Celery，构建一个可处理定时采集、异步任务调度和结果持久化的AI回答采集系统。适合AI开发者、后端工程师和架构师。前提：已开通阿里云账号，了解ECS、Redis、RDS基本使用。本文不涉及模型调用细节和前端展示。
业务任务与云上约束
采集系统需要定时向AI模型API发送问题，获取回答并存储。本地原型通常用单线程或简单队列，一旦任务量增大或进程崩溃，容易丢失任务或重复执行。云上部署必须解决几个实际问题：
任务调度：每天可能有数百个采集任务，需要按计划执行，且不能因为进程重启而丢失。
并发控制：模型API通常有频率限制，需要控制并发数，避免被限流。
数据持久化：采集结果需要可靠存储，支持后续查询和回溯。
监控与告警：任务失败、资源异常时需要及时发现。
成本控制：选择合理的资源规格，避免浪费。
环境和资源准备
云资源清单
资源 规格 用途 选择依据
ECS 2核4GB（按量付费） 运行采集主程序 预计日均任务数约1000，单个任务内存消耗约50MB，2核4GB可支撑4个并发Worker
Redis 256MB（标准版） 任务队列和缓存 任务队列长度通常不超过1000，256MB足够存储任务元数据
RDS MySQL 1核1GB（按量付费） 存储采集结果 每条回答约2KB，日均1000条，月数据量约60MB，1核1GB足够
日志服务 按量付费 采集日志和监控 日志量约500MB/月，按量付费成本可控
如果日均任务数超过10000，建议ECS升级到4核8GB，Redis使用512MB以上规格，RDS根据数据量调整存储空间。
环境配置
操作系统：Ubuntu 22.04 LTS
Python版本：3.10
依赖：requests, redis-py, pymysql, celery, dashscope
pip install requests redis pymysql celery dashscope
账号与权限
创建RAM用户，授予ECS、RDS、Redis和日志服务的读写权限。
创建AccessKey，配置到环境变量。
RDS和Redis设置白名单，仅允许ECS内网访问。
export ALIBABA_CLOUD_ACCESS_KEY_ID=""
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=""
export DASHSCOPE_API_KEY=""
方案对比与选择
方案 优点 缺点
单机多线程 实现简单 无法水平扩展，进程崩溃后任务丢失
Celery + Redis 成熟稳定，支持任务重试、定时调度和结果持久化 需要维护Worker进程
函数计算 + 消息队列 弹性伸缩，免运维 冷启动延迟，不适合长任务（如超过10分钟）
选择Celery + Redis方案，因为采集任务平均耗时约2-5秒，属于短任务，Celery的Worker模式可以稳定处理，且支持任务重试和定时调度。
核心实现
整体架构
flowchart LR
A[采集任务请求] --> B[Celery Beat定时调度]
B --> C[Redis任务队列]
C --> D[Celery Worker (ECS)]
D --> E[调用AI模型API]
E --> F[结果解析与清洗]
F --> G[RDS MySQL]
D --> H[日志服务]
H --> I[监控告警]
任务定义
tasks.py
from celery import Celery
import dashscope
import pymysql
import os
app = Celery('collector', broker='redis://:6379/0', backend='redis://:6379/1')
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def collect_answer(self, question, model='qwen-plus'):
try:
response = dashscope.Generation.call(
model=model,
prompt=question,
api_key=os.getenv('DASHSCOPE_API_KEY')
)
if response.status_code == 200:
answer = response.output.text
save_to_db(question, answer, model)
return {'question': question, 'status': 'success'}
else:
raise Exception(f"API error: {response.status_code}")
except Exception as exc:
raise self.retry(exc=exc)
def save_to_db(question, answer, model):
connection = pymysql.connect(
host='',
user='',
password='',
database='collector'
)
with connection.cursor() as cursor:
sql = "INSERT INTO answers (question, answer, model, created_at) VALUES (%s, %s, %s, NOW())"
cursor.execute(sql, (question, answer, model))
connection.commit()
connection.close()
定时调度
使用Celery Beat定期添加任务：
beat_schedule.py
from celery.schedules import crontab
from tasks import app
app.conf.beat_schedule = {
'collect-every-hour': {
'task': 'tasks.collect_answer',
'schedule': crontab(minute=0),
'args': ('最新AI技术趋势',)
},
}
启动Worker和Beat
启动Worker（建议使用supervisor管理）
celery -A tasks worker --loglevel=info --concurrency=4
启动Beat
celery -A tasks beat --loglevel=info
测试与监控结果
验证方式
任务执行后，检查RDS中answers表是否新增记录。
查看日志服务中Worker的日志，确认无异常。
模拟API返回非200状态码，观察任务是否自动重试。
正常情况下应当看到类似日志：
[2026-07-24 10:00:00,001: INFO/MainProcess] Task tasks.collect_answer succeeded in 2.34s: {'question': '最新AI技术趋势', 'status': 'success'}
监控指标说明
以下为预期目标值，实际表现取决于模型API响应速度和网络状况：
任务成功率：预期>99%（基于重试机制，临时故障可恢复）
平均响应时间：预期<3s（模型API平均响应约2s，加上网络和入库时间）
队列积压：预期<100（定时调度频率与Worker处理能力匹配）
成本、稳定性和安全分析
费用估算说明
以下费用为华东1地域按量付费的粗略估算，假设ECS和RDS每天运行24小时，Redis和日志服务按实际使用量计费。具体计费标准请以官方控制台为准。
资源 月费用（预估）
ECS（2核4GB按量） 约200元
Redis（256MB标准版） 约50元
RDS MySQL（1核1GB按量） 约100元
日志服务（按量） 约20元
总计 约370元
稳定性
Celery任务重试机制：max_retries=3，重试间隔60秒，可应对临时API故障。
Redis作为队列：Worker崩溃后任务不丢失，重启后继续消费。
RDS自动备份：防止数据丢失。
安全
AccessKey和API Key通过环境变量注入，不写死在代码。
RDS和Redis仅允许ECS内网访问，安全组配置最小权限。
日志脱敏：不记录完整API Key和数据库密码。
踩坑与适用边界
常见问题
任务重复执行：Celery Beat默认不保证幂等，需在业务层实现去重，例如在入库前检查相同question和time窗口内是否已存在记录。
API限流：设置合理的并发数（–concurrency=4）和重试延迟，避免触发模型API限流。如果仍被限流，可增加重试延迟或降低并发。
内存泄漏：长时间运行Worker，建议使用–max-tasks-per-child=1000参数，定期重启Worker进程。
适用边界
本方案适合中小规模采集（日均任务数<10000）。
大规模场景建议使用消息队列（如RocketMQ）和分布式Worker，并考虑使用容器化部署。
不适用于实时性要求极高的场景（秒级响应），因为Celery任务调度有最小延迟（约1秒）。
总结
从本地原型到云上部署，核心变化在于任务调度从单线程变为Celery异步队列，数据存储从本地文件变为RDS，并通过日志服务实现可观测。实际部署时，资源规格应根据任务量和数据量调整，同时注意API限流和任务幂等处理。本方案未涉及的部分包括：模型调用细节、前端展示、大规模分布式部署，这些可根据业务需求后续扩展。
