一、业务场景
在跨境电商的Web应用中，有些操作很耗时——发送邮件、生成报表、调用第三方API。如果在请求处理中同步执行这些操作，用户需要等待很长时间。
我最早用Flask写接口时，所有操作都是同步的：
python
@app.route('/api/order', methods=['POST'])def create_order(): # 1. 保存订单（50ms） order = save_order(request.json) # 2. 发送确认邮件（500ms） send_email(order.email, order.id) # 3. 调用第三方物流API（1000ms） call_shipping_api(order) # 4. 生成PDF回执（300ms） generate_pdf(order) return {'order_id': order.id}
总耗时1850ms，用户体验很差。
二、Celery + Redis的异步改造
python
from celery import Celeryimport time# 初始化Celeryapp = Celery( 'tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')@app.taskdef send_email_async(email, order_id): """异步发送邮件""" time.sleep(0.5) # 模拟发送邮件 print(f"邮件已发送到{email}，订单号{order_id}") return True@app.taskdef call_shipping_api_async(order_data): """异步调用物流API""" time.sleep(1.0) print(f"物流API调用成功，订单{order_data['id']}") return {'tracking_no': 'TRACK123'}@app.taskdef generate_pdf_async(order_id): """异步生成PDF""" time.sleep(0.3) print(f"PDF已生成，订单{order_id}") return f"/pdfs/{order_id}.pdf"
改造后的接口：
python
@app.route('/api/order', methods=['POST'])def create_order(): # 只有保存订单是同步的 order = save_order(request.json) # 其他操作全部异步 send_email_async.delay(order.email, order.id) call_shipping_api_async.delay(order.to_dict()) generate_pdf_async.delay(order.id) # 立即返回，总耗时<100ms return {'order_id': order.id, 'status': 'processing'}
三、任务状态查询
用户需要知道异步任务的状态：
python
from celery.result import AsyncResult@app.route('/api/order/status/')def get_order_status(order_id): # 从数据库获取订单状态 order = get_order(order_id) return { 'order_id': order_id, 'status': order.status, 'email_sent': order.email_sent, 'shipping_ready': order.shipping_ready, 'pdf_url': order.pdf_url }
四、任务重试与错误处理
python
@app.task(bind=True, max_retries=3, default_retry_delay=60)def call_shipping_api_with_retry(self, order_data): try: response = requests.post( 'https://shipping-api.example.com/create', json=order_data, timeout=5 ) if response.status_code != 200: raise Exception(f"API返回{response.status_code}") return response.json() except Exception as e: # 重试 self.retry(exc=e)
五、定时任务
Celery还支持定时任务（Beat）：
python
from celery.schedules import crontabapp.conf.beat_schedule = { 'check_pending_orders': { 'task': 'tasks.check_pending_orders', 'schedule': crontab(minute='*/5'), # 每5分钟执行 }, 'sync_inventory_daily': { 'task': 'tasks.sync_inventory', 'schedule': crontab(hour=2, minute=0), # 每天凌晨2点 }}@app.taskdef check_pending_orders(): """检查未支付订单，超时自动取消""" pending = get_pending_orders(timeout=30) # 30分钟未支付 for order in pending: cancel_order(order.id) send_cancel_notification(order.user_id)
六、监控与告警
python
from celery.events import EventReceiverdef monitor_celery_tasks(): """监控Celery任务""" with app.connection() as connection: recv = EventReceiver(connection, handlers={ 'task-failed': handle_task_failed, 'task-succeeded': handle_task_succeeded, }) recv.capture(limit=None, timeout=None)def handle_task_failed(event): """任务失败时发送告警""" print(f"任务失败: {event['uuid']}, 错误: {event['exception']}") # 发送告警到钉钉 send_alert(f"Celery任务失败: {event['name']}")
七、总结
Flask + Celery的异步改造让接口响应时间从秒级降到了毫秒级。核心经验是：同步只做必要操作，耗时操作全部异步化，用任务状态查询接口让用户感知进度。
