Django 2.2项目如何用Python async函数替代少量Celery后台任务

实现步骤 #

1. 依赖安装 #

你需要先安装asgiref包用来做同步/异步逻辑互转，如果有延迟触发任务的需求，可以安装适配Python 3.6的APScheduler 3.9.1版本。

2. 任务定义改造 #

直接把原来带@shared_task装饰器的同步函数改成async函数即可，原有业务逻辑可以基本保留：

# tasks.py 改造后示例
import asyncio
from asgiref.sync import async_to_sync
from backend.tasks import json_result

# 全局复用一个事件循环即可，不要重复创建
loop = asyncio.get_event_loop()

@json_result
async def task1(args):
    # 原有业务逻辑保持不变
    # 若逻辑内有IO操作，建议替换为异步库（比如发信用aiosmtplib，发HTTP请求用aiohttp）性能更好
    ...

async def send_email(user_pk):
    # 原有发邮件逻辑
    ...

async def task2(args):
    # 原有业务逻辑
    ...
    # 原Celery的countdown=10延迟触发写法替换为下面的实现
    loop.call_later(10, asyncio.create_task, send_email(user.pk))

3. 任务调用方式 #

Django 2.2默认所有视图、业务逻辑都是同步的，不同场景调用方式不同：

• 不需要等待任务返回结果，直接后台运行：

# 任意同步代码位置调用，任务会在事件循环后台执行，不会阻塞当前主线程
loop.create_task(task1(your_args))

• 需要等待任务执行完成拿到结果再往下走：

from asgiref.sync import async_to_sync
result = async_to_sync(task1)(your_args)

后续如果你升级Django到3.2及以上版本，原生支持异步视图，就可以直接用await task1(your_args)的方式调用。

更简单的替代方案 #

如果不想接触async的事件循环、同步异步互转等坑，还有更轻量的方案适配你的低负载场景：用django-background-tasks库，不需要单独部署Broker、worker，任务存在数据库中，只需要跑一个轻量进程就可以执行，支持延迟触发、任务重试等基础能力，写法和原Celery任务几乎一致，改造成本极低。

注意事项 #

• Python 3.6的asyncio存在部分已知bug，如果运行中遇到事件循环相关异常，建议升级Python到3.7+，Django 2.2完全兼容Python 3.7
• 异步函数内不要直接调用大量Django同步ORM操作，会阻塞事件循环，低负载场景可以临时凑合用，高负载场景建议用异步ORM包装
• 不要在异步任务中运行CPU密集型逻辑，会卡住整个事件循环，导致所有异步任务都无法执行

学习资源 #

• Django官方异步支持文档
• Python官方asyncio模块文档
• 《Python异步编程实战》入门书籍

内容的提问来源于stack exchange，提问作者Somraj Chowdhury