Python如何实现不阻塞返回的异步线程 解决Django推送延迟问题

核心需求：将Firebase推送逻辑从请求链路中剥离，执行推送时不阻塞接口返回，后台自动完成推送任务。

问题复现 #

原有同步视图逻辑如下，Firebase接口延迟会直接拖慢整个接口响应速度：

def post(View):
    # 校验帖子参数
    # 写入数据库
    # 同步调用Firebase发送推送（阻塞点）
    # 返回响应

可选实现方案 #

方案1：线程池异步执行（兼容WSGI部署，改造成本最低） #

适合老项目不想调整部署架构的场景，用Python内置线程池跑IO密集型的推送任务，无需等待执行结果：

from concurrent.futures import ThreadPoolExecutor
# 初始化线程池，IO密集型任务可设置稍大的线程数
executor = ThreadPoolExecutor(max_workers=10)

# 封装独立的推送函数
def send_firebase_push(post_content):
    # 原有Firebase推送逻辑
    pass

def post(View):
    # validate post ect...
    # store it in database
    # 提交推送任务到线程池，无需等待返回
    executor.submit(send_firebase_push, post_content=your_post_data)
    # 直接返回响应
    return Response()

注意点：线程池大小根据业务推送量调整，避免线程过多占用服务器资源。

方案2：Django原生异步视图（Django>=3.1，ASGI部署场景） #

新项目采用ASGI部署的情况下，可直接用Python asyncio的后台任务能力：

import asyncio
from django.http import JsonResponse

# 异步推送函数
async def send_firebase_push(post_content):
    # 异步Firebase推送逻辑（需用异步客户端）
    pass

async def post(request):
    # 校验参数、写入数据库（支持同步ORM，Django会自动处理线程切换）
    # 创建后台任务执行推送，无需等待
    asyncio.create_task(send_firebase_push(your_post_data))
    return JsonResponse({"code":0, "msg":"发布成功"})

注意点：需采用Uvicorn、Daphne等ASGI服务器部署，不能用传统WSGI服务器部署。

方案3：Celery任务队列（生产级高可用方案） #

适合中大型项目、推送量较大、需要推送失败重试、执行记录的场景：

1. 先安装依赖：pip install celery redis
2. 配置Celery和Django集成，新建tasks.py：

from celery import shared_task

@shared_task(bind=True, max_retries=3)
def send_firebase_push(self, post_content):
    try:
        # 原有Firebase推送逻辑
        pass
    except Exception as e:
        # 推送失败自动重试
        self.retry(exc=e, countdown=5)

3. 视图中调用：

from .tasks import send_firebase_push

def post(View):
    # validate post ect...
    # store it in database
    # 提交异步任务，立即返回
    send_firebase_push.delay(your_post_data)
    return Response()

注意点：需要单独部署Celery worker进程，采用Redis或RabbitMQ作为任务队列中间件。

选型建议 #

• 小型项目、推送量低：优先选方案1，改造成本极低，无额外依赖
• 新项目ASGI部署：选方案2，符合Django未来技术演进方向
• 中大型项目、对推送可靠性有要求：选方案3，支持任务监控、重试、失败回溯

内容的提问来源于stack exchange，提问作者Ben Owen