如何在Django接口中实现异步任务的串行执行以保证文件写入的顺序性？

问题分析 #

你当前的代码核心问题在于使用run_in_executor将任务提交到了多线程池，默认线程池的线程数等于CPU核心数，多个请求的任务会并行执行，直接导致文件写入顺序混乱。另外，Django同步视图中每个请求可能复用或创建不同的事件循环，无法保证所有任务绑定到同一个串行执行的上下文里。

你的需求是：请求可以立即返回响应，文件写入任务按请求到达顺序串行执行，且任务提交方无需感知队列状态。下面分两种部署场景给出针对性解决方案：

方案一：单进程部署轻量方案（进程内串行队列） #

如果你的Django是单进程部署（比如用runserver或gunicorn --workers=1），可以通过全局任务队列+单独消费线程实现串行执行，无需引入额外依赖。

1. 初始化全局队列与消费线程

在你的Django app的apps.py中，初始化全局任务队列并启动守护线程消费队列：

import asyncio
import threading
import logging
from django.apps import AppConfig

logger = logging.getLogger(__name__)

# 全局任务队列，所有写入任务都会被放到这里
task_queue = asyncio.Queue()

async def process_task_queue():
    """持续消费队列，逐个执行任务"""
    while True:
        task = await task_queue.get()
        try:
            # 将同步任务转为异步执行，不阻塞事件循环
            await asyncio.to_thread(task)
        except Exception as e:
            logger.error(f"任务执行失败: {e}")
        finally:
            task_queue.task_done()

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name'

    def ready(self):
        # 启动消费守护线程，随Django进程退出自动终止
        def start_consumer():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.run_until_complete(process_task_queue())

        threading.Thread(target=start_consumer, daemon=True).start()

2. 修改视图代码，将任务加入队列

在视图中，不再直接调用run_in_executor，而是把写入任务放到全局队列：

from django.http import Response
from rest_framework import status
from django.core.files.storage import default_storage
import logging
from tenacity import retry, stop_after_attempt, wait_exponential
from .apps import task_queue

logger = logging.getLogger(__name__)

class YourFileAppendView(APIView):
    def post(self, request, *args, **kwargs):
        # 生成请求唯一标识，用于日志追踪
        request_number = request.META.get('HTTP_X_REQUEST_ID', f"req-{id(request)}")
        # 替换为你实际要写入的内容（对应原代码中的append变量）
        append_content = b"your_content_here"
        file_name = "target_file.txt"

        def append_to_file():
            """带重试逻辑的文件写入任务"""
            @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=3, max=10))
            def _append_with_retry():
                with default_storage.open(file_name, "a+b") as f:
                    logger.info(f"开始写入 @ {request_number}")
                    f.write(b"\n" + append_content)
                    logger.info(f"完成写入 @ {request_number}")
            
            try:
                _append_with_retry()
            except Exception as e:
                logger.error(f"写入失败（已重试3次）@ {request_number}: {str(e)}")

        # 将任务加入全局队列，消费线程会按顺序执行
        task_queue.put_nowait(append_to_file)

        logger.info(f"返回响应 @ {request_number}")
        return Response(None, status=status.HTTP_200_OK)

方案二：多进程/分布式部署方案（Celery串行队列） #

如果你的Django是多进程或多机器部署，进程内的全局队列无法跨进程共享，这时候需要用**分布式任务队列（Celery）**实现全局串行执行。

1. 安装并配置Celery

首先安装依赖：

pip install celery redis

在项目根目录创建celery.py配置文件：

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

在settings.py中添加Celery配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Redis作为消息中间件
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

2. 创建串行执行的Celery任务

在app的tasks.py中定义任务：

import logging
from celery import shared_task
from django.core.files.storage import default_storage
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger(__name__)

@shared_task(queue='serial_file_write')  # 指定专属串行队列
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=3, max=10))
def append_to_file_task(request_number, append_content, file_name):
    with default_storage.open(file_name, "a+b") as f:
        logger.info(f"开始写入 @ {request_number}")
        f.write(b"\n" + append_content)
        logger.info(f"完成写入 @ {request_number}")

3. 修改视图调用任务

from django.http import Response
from rest_framework import status
import logging
from .tasks import append_to_file_task

logger = logging.getLogger(__name__)

class YourFileAppendView(APIView):
    def post(self, request, *args, **kwargs):
        request_number = request.META.get('HTTP_X_REQUEST_ID', f"req-{id(request)}")
        append_content = b"your_content_here"
        file_name = "target_file.txt"

        # 将任务发送到Celery串行队列
        append_to_file_task.delay(request_number, append_content, file_name)

        logger.info(f"返回响应 @ {request_number}")
        return Response(None, status=status.HTTP_200_OK)

4. 启动串行执行的Celery Worker

启动Worker时指定队列，并设置concurrency=1（仅允许一个线程执行任务，保证串行）：

celery -A your_project worker -Q serial_file_write --concurrency=1 --loglevel=info

关键注意事项 #

1. 单进程方案限制：仅适用于单进程部署，多进程下每个进程有独立队列，无法实现全局串行。
2. Celery方案扩展性：支持多进程、多机器部署，Redis维护全局任务队列，Worker严格按顺序执行任务。
3. 重试逻辑影响：两种方案的重试逻辑都在任务内部执行，失败重试时后续任务会等待当前任务完成，不会打乱整体任务顺序。

内容来源于stack exchange