初学Django与Celery，如何将Celery任务全部结果存入单个文本文件？

现有views.py的调用逻辑不需要修改，仅需调整tasks.py的配置即可实现需求，核心要解决两个问题：路径正确性、多进程并发写文件的冲突。

第一步：调整tasks.py的基础导入和公共配置 #

在tasks.py开头添加以下导入和配置：

import os
import datetime
import requests
from celery import shared_task  # 如果你用的是shared_task而不是app.task就保留，否则替换成你的app实例导入
from django.conf import settings

# 兼容不同操作系统的文件锁
try:
    import fcntl  # Linux/macOS系统使用
except ImportError:
    import msvcrt  # Windows系统使用

# 统一结果文件的绝对路径，避免相对路径导致的位置错误
RESULT_FILE_PATH = os.path.join(settings.BASE_DIR, "celery_task_results.txt")

第二步：选择结果存储实现方式 #

方案1：通用信号方案（推荐） #

用Celery自带的任务完成信号捕获所有任务的返回值，自动写入文件，后续新增的Celery任务不需要单独修改代码即可自动存储结果。
在tasks.py中添加信号监听逻辑：

from celery.signals import task_postrun

@task_postrun.connect
def write_result_to_file(sender=None, task_id=None, args=None, kwargs=None, retval=None, state=None, **extra):
    # 仅存储执行成功的任务结果，需要存储失败结果可以去掉state判断
    if state == "SUCCESS":
        # 自定义写入的内容格式，可根据需求调整
        log_content = (
            f"任务ID：{task_id}\n"
            f"任务名称：{sender.name}\n"
            f"请求参数：{args} {kwargs}\n"
            f"返回结果：{retval}\n"
            f"执行时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
            f"{'-'*30}\n"
        )
        # 加锁写入避免并发冲突
        with open(RESULT_FILE_PATH, "a", encoding="utf-8") as f:
            # 加排他锁，同一时间仅允许一个进程写入
            try:
                fcntl.flock(f, fcntl.LOCK_EX)
            except NameError:
                msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
            f.write(log_content)
            f.flush()  # 强制将缓存内容写入磁盘，避免丢失

调整你原有的process任务，只需要保持返回结果即可，不需要加写入逻辑：

@app.task
def process(url):
    r = requests.get(url, allow_redirects=True)
    # 可根据需求返回r.text（文本内容）、r.content（二进制内容）或其他自定义数据
    return r.text

方案2：单任务自定义方案 #

如果只想存储process这一个任务的结果，其他任务不需要存储，可以直接把写入逻辑加到process任务内部：

@app.task
def process(url):
    r = requests.get(url, allow_redirects=True)
    task_result = r.text
    log_content = (
        f"请求URL：{url}\n"
        f"响应内容：{task_result}\n"
        f"执行时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
        f"{'-'*30}\n"
    )
    # 加锁写入文件
    with open(RESULT_FILE_PATH, "a", encoding="utf-8") as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX)
        except NameError:
            msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
        f.write(log_content)
        f.flush()

常见注意事项 #

• 不要使用相对路径写文件，否则Celery worker的启动目录和项目根目录不一致时，文件会生成在非预期位置。
• 必须保留文件锁逻辑，否则多个Celery worker同时写入同一个文件时，会出现内容交错、覆盖丢失的问题。
• 请提前确认Celery已经正确加载Django配置，否则会读取不到settings.BASE_DIR，启动worker时报错。
• 纯文本存储仅适合开发测试场景，生产环境建议使用Celery自带的结果后端（Redis、数据库等）存储任务结果，需要时再导出到文本文件。

额外提示：你现有代码里的urls=["https//:.....com,https//:.....com,etc.."]写法存在小问题，当前列表仅包含1个拼接了多个URL的字符串，循环时只会触发1次任务，如果要遍历多个URL，需要写成urls=["https://xxx.com", "https://yyy.com"]这种每个URL单独作为列表元素的格式。

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