如何在Django中终止长请求并停止处理流程？

方案一：中间件+信号实现请求超时监控 #

• 核心逻辑：请求进入时启动后台线程监控时长，超时后触发信号终止请求流程，视图捕获信号抛出的异常并返回错误。
• 实现步骤：
  1. 定义超时通知信号：

     from django.dispatch import Signal
     
     request_timeout = Signal()
     

  2. 编写超时监控中间件：

     import threading
     from django.http import HttpResponseServerError
     
     class TimeoutMiddleware:
         def __init__(self, get_response):
             self.get_response = get_response
     
         def __call__(self, request):
             timeout_flag = threading.Event()
     
             def monitor():
                 # 设置10秒超时阈值
                 timeout_flag.wait(10)
                 if not timeout_flag.is_set():
                     request_timeout.send(sender=self.__class__, request=request)
     
             # 启动守护线程执行监控
             monitor_thread = threading.Thread(target=monitor, daemon=True)
             monitor_thread.start()
     
             try:
                 response = self.get_response(request)
             except Exception as e:
                 if str(e) == "RequestTimeout":
                     return HttpResponseServerError("请求超时，请稍后重试")
                 raise
             finally:
                 # 请求完成，终止监控线程
                 timeout_flag.set()
     
             return response
     

  3. 在目标视图中绑定信号处理逻辑：

     from django.shortcuts import render
     from .signals import request_timeout
     import time
     
     def long_running_view(request):
         def handle_timeout(sender, **kwargs):
             raise Exception("RequestTimeout")
     
         # 绑定信号（避免弱引用自动回收）
         request_timeout.connect(handle_timeout, sender=TimeoutMiddleware, weak=False)
     
         try:
             # 模拟耗时业务逻辑
             time.sleep(15)
             return render(request, 'success.html')
         finally:
             # 解绑信号
             request_timeout.disconnect(handle_timeout, sender=TimeoutMiddleware)
     

方案二：Celery异步任务+状态查询 #

• 核心逻辑：把耗时任务交给Celery异步执行，前端通过轮询或WebSocket查询任务状态，超过10秒则主动终止任务并返回超时提示。
• 实现步骤：
  1. 定义Celery异步任务：

     from celery import shared_task
     import time
     
     @shared_task(bind=True)
     def long_task(self):
         # 模拟耗时操作
         time.sleep(15)
         return "任务执行完成"
     

  2. 编写启动任务的视图：

     from django.http import JsonResponse
     from .tasks import long_task
     
     def start_task(request):
         task = long_task.delay()
         return JsonResponse({"task_id": task.id})
     

  3. 编写任务状态查询视图，加入超时判断：

     from celery.result import AsyncResult
     from django.http import JsonResponse
     import time
     
     def check_task(request, task_id):
         result = AsyncResult(task_id)
         # 计算任务运行时长
         run_duration = time.time() - float(result.task_id.split('-')[0])
         if run_duration > 10:
             # 强制终止任务
             result.revoke(terminate=True)
             return JsonResponse({"status": "timeout", "msg": "请求超时，请稍后重试"})
         if result.ready():
             return JsonResponse({"status": "success", "data": result.get()})
         else:
             return JsonResponse({"status": "running"})
     

方案三：Gunicorn服务器层面超时控制 #

• 核心逻辑：利用WSGI服务器的超时配置，强制终止超时时长的请求，同时配置Django自定义500错误页返回超时提示。
• 实现步骤：
  1. 启动Gunicorn时设置超时参数：

     gunicorn your_project.wsgi:application --timeout 10
     

  2. 配置Django自定义500错误视图：

     # views.py
     def custom_500(request):
         return HttpResponseServerError("请求超时，请稍后重试", status=500)
     
     # urls.py
     handler500 = 'your_app.views.custom_500'
     

• 注意：此方案是全局统一控制，无法针对单个视图做差异化配置。

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