如何让Django视图返回响应后执行长耗时网页爬取任务？

这个问题太典型了——同步视图里跑长任务肯定会阻塞请求，用户得等半天才能收到响应，体验极差。咱们有几种靠谱的办法解决，从简单临时方案到生产级方案都有，我一个个给你拆解：

方案1：用Python线程（快速临时方案，不推荐生产环境） #

如果你只是想快速验证功能，不想折腾复杂工具，可以用Python内置的threading模块把长任务丢到后台线程。这种方式最省事，但缺点也很明显：任务没有持久化，服务器重启或进程被回收时，未完成的任务会直接中断，也没法追踪任务状态。

修改后的视图代码如下：

import threading
from django.http import JsonResponse
from django.utils import timezone
import requests

def my_view(request):
    batch = BatchTable.objects.create(user=request.user)
    # 启动独立线程执行长任务，不阻塞当前请求
    threading.Thread(target=long_running_operation, args=(batch,)).start()
    return JsonResponse({'success': 'True', 'message': 'Batch started'})

def long_running_operation(batch):
    try:
        list_response = requests.get('https://agiven.web.service/rest/api/objects')
        object_list = list_response.json()
        batch.total = len(object_list)
        batch.save()
        
        for object_entry in object_list:
            object_response = requests.get(f'https://agiven.web.service/rest/api/objects/{object_entry["id"]}')
            object_dict = object_response.json()
            object_dict['batch'] = batch
            ObjectTable.objects.create(**object_dict)
            batch.progress += 1
            batch.save()
        batch.finalized = timezone.now()
        batch.save()
    except Exception as ex:
        batch.fail = str(ex)
        batch.failed = timezone.now()
        batch.save()

方案2：用Django异步视图（Django 3.1+适用） #

如果你用的是较新版本的Django，异步视图是个不错的轻量选择，不需要额外引入任务队列工具。但要注意，原来的同步IO操作（比如requests调用）得改成异步的，这里我们用aiohttp替代requests。

1. 先安装依赖：

pip install aiohttp

2. 修改代码为异步版本：

import asyncio
import aiohttp
from django.http import JsonResponse
from django.utils import timezone

async def my_view(request):
    batch = BatchTable.objects.create(user=request.user)
    # 把长任务丢到后台异步执行，不阻塞响应返回
    asyncio.create_task(long_running_operation(batch))
    return JsonResponse({'success': 'True', 'message': 'Batch started'})

async def long_running_operation(batch):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get('https://agiven.web.service/rest/api/objects') as list_response:
                object_list = await list_response.json()
                batch.total = len(object_list)
                batch.save()
                
                for object_entry in object_list:
                    async with session.get(f'https://agiven.web.service/rest/api/objects/{object_entry["id"]}') as object_response:
                        object_dict = await object_response.json()
                        object_dict['batch'] = batch
                        ObjectTable.objects.create(**object_dict)
                        batch.progress += 1
                        batch.save()
                batch.finalized = timezone.now()
                batch.save()
    except Exception as ex:
        batch.fail = str(ex)
        batch.failed = timezone.now()
        batch.save()

⚠️ 注意：生产环境必须用ASGI服务器（比如Uvicorn）部署，传统的WSGI服务器（比如Gunicorn）会把异步视图当成同步视图执行，起不到异步效果。

方案3：用Celery（生产环境首选） #

如果你的项目需要处理大量异步任务、支持任务重试、状态追踪，Celery是工业级的解决方案。它通过消息队列（比如Redis或RabbitMQ）持久化任务，就算服务器重启，Worker进程重启后还能继续执行未完成的任务。

步骤1：安装依赖 #

pip install celery redis

步骤2：配置Celery #

在你的Django项目根目录（和settings.py同级）创建celery.py：

import os
from celery import Celery

# 设置Django配置文件环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

app = Celery('你的项目名')
# 从Django配置中读取Celery配置，前缀为CELERY_
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有app下的tasks.py文件
app.autodiscover_tasks()

在settings.py中添加Celery配置：

# 消息队列地址（这里用Redis）
CELERY_BROKER_URL = 'redis://localhost:6379/0'
# 任务结果存储地址
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

步骤3：把长任务改成Celery任务 #

在你的app下创建tasks.py：

from celery import shared_task
from django.utils import timezone
import requests
from .models import BatchTable, ObjectTable

@shared_task
def long_running_operation(batch_id):
    batch = BatchTable.objects.get(id=batch_id)
    try:
        list_response = requests.get('https://agiven.web.service/rest/api/objects')
        object_list = list_response.json()
        batch.total = len(object_list)
        batch.save()
        
        for object_entry in object_list:
            object_response = requests.get(f'https://agiven.web.service/rest/api/objects/{object_entry["id"]}')
            object_dict = object_response.json()
            object_dict['batch'] = batch
            ObjectTable.objects.create(**object_dict)
            batch.progress += 1
            batch.save()
        batch.finalized = timezone.now()
        batch.save()
    except Exception as ex:
        batch.fail = str(ex)
        batch.failed = timezone.now()
        batch.save()

步骤4：修改视图调用异步任务 #

from django.http import JsonResponse
from .tasks import long_running_operation

def my_view(request):
    batch = BatchTable.objects.create(user=request.user)
    # 异步提交任务到Celery队列，立即返回响应
    long_running_operation.delay(batch.id)
    return JsonResponse({'success': 'True', 'message': 'Batch started'})

步骤5：启动Celery Worker #

在终端运行：

celery -A 你的项目名 worker --loglevel=info

这个方案稳定性最强，适合生产环境长期使用，还能扩展任务重试、定时任务等功能。

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