下载任务怎么排队：Celery + Redis 的任务状态机、进度回传与失败重试

最早版本的解析功能是这么写的：用户在页面上贴一个链接，点"解析"，Django 视图里直接调用解析逻辑，跑完返回结果。本地测试一切正常——本地解析一个 B 站链接大概两秒。

上线第一天就炸了。有个用户贴了个有 300 多个分集的合集链接，请求跑了 40 多秒，gunicorn 的 worker 被占住不动；同一时间又来了几个请求，剩下的 worker 全被占满，整个站点响应变慢到打不开。更尴尬的是 nginx 超时之后给用户返回了 504，但后台的任务还在跑——worker 干着没人会来取的活。

把耗时任务挪到 Celery 之后，504 没了，但新问题一个接一个冒出来：前端拿不到进度（一直显示"解析中"）；同一个任务被执行了两次；Redis 内存涨到 8G；每天凌晨的清理任务偶尔跑两次。这篇文章就是这一路的记录。

TL;DR：耗时任务必须出请求，用 Celery + Redis 解耦。四个必须做对的事：任务状态机要自己定义（内置状态不够用，进度要靠 update_state 回传）；重试必须配幂等（acks_late + 幂等键，否则网络抖动会让你重复下载一遍 4GB 的文件）；长任务要关掉预取（prefetch_multiplier=1）并且 visibility_timeout 要大于任务最长耗时；结果后端和 beat 都要做清理，不然数据库和 Redis 会慢慢被撑爆。

一、为什么不能在请求里干活
二、整体架构长什么样
三、最小可用配置（Django + Celery + Redis）
四、任务状态机：内置状态不够用
五、进度回传：从"一直在转圈"到真实百分比
六、重试与幂等：别把 4GB 的文件下两遍
七、并发、预取与资源控制
八、定时任务 beat：重复触发与时区坑
九、Redis 与结果后端的运维
十、我踩过的坑清单

一、为什么不能在请求里干活

先看那个 504 是怎么发生的。gunicorn 默认同步 worker，一个 worker 同时只能处理一个请求。假设 4 个 worker：

请求 1（解析合集）→ worker A，跑 40 秒
请求 2、3、4、5  → 占满 worker B/C/D，剩下 1 个排队
请求 6+          → 全部排队，站点"打不开"

nginx 在 30 秒（或你配的超时）后断开连接返回 504，但 Django 那边不知道客户端已经走了，任务继续跑完。也就是说：

用户看到失败；
服务器白干了；
期间整个站点的吞吐被拖垮。

换个角度想：这类任务的本质是"开始一次长时间的工作，然后告诉用户进展"，根本不是一个"请求-响应"模型。硬塞进 HTTP 请求里，等于把一个异步问题用同步方式解决，迟早出事。

判断标准很简单：任何可能超过 2 秒的操作，都别放在请求里。解析、下载、转码、批量处理、调用外部慢 API，全算。

二、整体架构长什么样

我们现在的架构（规模不大，但跑得很稳）：

┌──────────┐   1. 提交     ┌─────────────┐
│  浏览器   │ ───────────► │ Django 视图  │
│          │ ◄─────────── │ (立即返回 id) │
└──────────┘   2. task_id  └─────────────┘
     │                            │ 3. delay()
     │ 4. SSE / 轮询              ▼
     │                     ┌──────────────┐
     └───────────────────► │ Redis broker │
          (进度事件)        └──────────────┘
                                  │ 5. 消费
                                  ▼
                          ┌──────────────┐
                          │ Celery worker │
                          └──────────────┘
                                  │ 6. 写状态
                                  ▼
                    ┌──────────────────────────┐
                    │ Django DB (DownloadTask) │
                    │ + celery results 表       │
                    └──────────────────────────┘

几个设计选择，说下理由：

为什么用 Redis 当 broker 而不是 RabbitMQ？ 规模小、部署简单，而且我们本来就要用 Redis 做缓存和 SSE 发布。代价是：Redis 的消息可靠性不如 RabbitMQ（不做持久化配置的话，Redis 重启会丢消息）。对我们这种"解析失败大不了重来一次"的场景可以接受；如果是转账、计费这类任务，老实上 RabbitMQ。

为什么进度走数据库 + SSE 而不是直接读 Celery 的结果后端？ 因为我们有自己的 DownloadTask 模型，任务状态要跟业务字段（URL、文件名、大小、错误信息）放在一起，前端一次查询能拿到所有东西。Celery 的结果后端只存 Celery 自己的元数据。前端通过 SSE 订阅 /task/<id>/events/，模型保存后发信号往 Redis 频道 publish，SSE 视图把事件推给浏览器。

为什么不一开始就用 WebSocket？ SSE 够用了（单向推送），实现比 WebSocket 简单，nginx 反代也好配。后面真需要双向再换。

三、最小可用配置（Django + Celery + Redis）

项目结构：

video_downloader/
├── __init__.py          # 在这里 import celery app
├── celery.py            # Celery 实例
├── settings.py
└── ...
downloader/
├── tasks.py             # @shared_task
└── models.py            # DownloadTask

video_downloader/celery.py：

import os
from celery import Celery

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

app = Celery('video_downloader')
# 从 Django settings 里读所有 CELERY_ 开头的配置
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

video_downloader/__init__.py：

from .celery import app as celery_app

__all__ = ('celery_app',)

这行 import 不能忘，忘了的话 worker 启动时报 ImportError: No module named celery_app 或者干脆找不到任务，我第一次配的时候卡在这儿十几分钟。

settings.py 的关键配置：

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'django-db'          # 用 django-celery-results
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = TIME_ZONE

# 长任务必须改的两项
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'visibility_timeout': 43200,   # 默认 3600，长任务必须调大
}
CELERY_WORKER_PREFETCH_MULTIPLIER = 1        # 长任务关掉预取
CELERY_TASK_ACKS_LATE = True

# 队列拆分
CELERY_TASK_ROUTES = {
    'downloader.tasks.fetch_info': {'queue': 'parse'},
    'downloader.tasks.download_video': {'queue': 'download'},
    'downloader.tasks.transcode': {'queue': 'transcode'},
}

视图里提交任务：

from .tasks import fetch_info

def submit(request):
    url = request.POST['url']
    task = DownloadTask.objects.create(url=url, status='PENDING')
    # 把业务主键传进去，方便任务里更新状态
    fetch_info.apply_async(args=[task.id], queue='parse')
    return JsonResponse({'task_id': task.id})

启动 worker（supervisor 里我们是这么配的）：

celery -A video_downloader worker --loglevel=info --concurrency=4 \
  --without-gossip --without-mingle --without-heartbeat

--without-gossip --without-mingle --without-heartbeat 是我们后来加的。这三个是 worker 之间的通信机制，在多 worker 场景下有用，但我们只有一个 worker 进程，关掉能少一堆无谓的日志和 Redis 通信。

四、任务状态机：内置状态不够用

Celery 内置状态是：PENDING → STARTED → SUCCESS / FAILURE / RETRY / REVOKED。

这套状态表达不了我们要的东西。用户想知道的是"现在在干嘛"：正在解析地址？正在下载第 3 个分片？正在合并？所以我在业务模型里自己定义了一套：

状态	含义	前端展示
`PENDING`	已入队，还没被 worker 拿到	排队中
`PARSING`	正在解析页面/接口	正在解析…
`DOWNLOADING`	正在下载	正在下载 45%
`MERGING`	分片合并 / 转码中	正在合并…
`SUCCESS`	完成	下载/预览
`FAILED`	失败（带错误分类）	失败：链接已失效
`CANCELED`	用户取消或超时终止	已取消

关键原则：业务状态写在自己的模型里，不要只依赖 Celery 的状态。 原因有三个：

Celery 的状态在 Redis/结果表重启后可能丢失，你的业务数据不能跟着丢；
业务状态要跟业务字段（文件名、大小、错误原因）一起更新，一次 save() 搞定；
前端只查你的模型，不用理解 Celery 的内部状态。

任务里更新状态：

@shared_task(bind=True)
def download_video(self, task_id):
    task = DownloadTask.objects.get(id=task_id)
    task.status = 'DOWNLOADING'
    task.save(update_fields=['status', 'updated_at'])
    ...

update_fields 是个好习惯：Django 默认 save() 会把所有字段写一遍，并发更新时容易把别的字段覆盖回去。

状态流转要显式检查，别让一个已经取消的任务继续往下跑：

def _check_canceled(task_id):
    """任务可能已经被用户取消，在每个耗时步骤前检查一次"""
    fresh = DownloadTask.objects.filter(id=task_id).values('status').first()
    return fresh and fresh['status'] == 'CANCELED'

这是因为 task.revoke() 只能撤销还没开始执行的任务，已经开始跑的不会自己停。想真正中断一个正在跑的任务，只能让它自己定期检查"我是不是该停了"。我一般把检查放在每个分片下载的循环里，代价可以忽略。

五、进度回传：从"一直在转圈"到真实百分比

第一版上线后最常见的反馈是："点了之后一直转圈，我根本不知道它在干嘛，也不知道要等多久。"

进度回传有两个层面：Celery 层（update_state）和业务层（写数据库 + SSE）。

Celery 层的进度

@shared_task(bind=True)
def download_video(self, task_id, total):
    for i in range(total):
        do_download_chunk(i)
        self.update_state(
            state='PROGRESS',
            meta={'current': i + 1, 'total': total,
                  'percent': round((i + 1) / total * 100, 1)},
        )

前端拿进度：

res = download_video.AsyncResult(task_id)
res.state          # 'PROGRESS'
res.info           # {'current': 3, 'total': 100, 'percent': 3.0}

注意：PROGRESS 是自定义状态，Celery 不认识它，但允许你用。它会出现在 res.state 里，res.info 就是你塞的 meta。

有个坑：update_state 的写入频率要控制。我一开始每下载一个分片就写一次，一个 800 分片的视频写了 800 次 Redis，日志里全是 Task ... updated。后来改成"每 2% 或者每 2 秒写一次"：

last = 0
for i in range(total):
    do_download_chunk(i)
    percent = (i + 1) / total * 100
    if percent - last >= 2:
        self.update_state(state='PROGRESS', meta={'percent': round(percent, 1)})
        last = percent

业务层的实时推送

Celery 的进度要前端主动查才拿得到，用户体感还是"卡"。我们最后用的是 SSE：模型保存后发信号，往 Redis 频道 publish。

# downloader/realtime.py（简化版）
import json
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.cache import cache
from .models import DownloadTask

CHANNEL = 'task_events'


def publish(event: dict):
    # 用 cache 的 redis 客户端直接 publish
    client = cache.client.get_client(write=True)
    client.publish(CHANNEL, json.dumps(event))


@receiver(post_save, sender=DownloadTask)
def on_task_saved(sender, instance, **kwargs):
    publish({
        'task_id': instance.id,
        'status': instance.status,
        'percent': instance.percent,
        'message': instance.message,
    })

视图端（用 StreamingHttpResponse 推 SSE）：

import json
import time
from django.http import StreamingHttpResponse
from django.core.cache import cache


def task_events(request, task_id):
    def stream():
        client = cache.client.get_client(write=True)
        pubsub = client.pubsub()
        pubsub.subscribe('task_events')
        yield 'retry: 3000\n\n'          # 断线重连间隔
        last_ping = time.time()
        try:
            for message in pubsub.listen():
                if message['type'] != 'message':
                    # 心跳，防止 nginx / 浏览器超时断开
                    if time.time() - last_ping > 15:
                        yield ': ping\n\n'
                        last_ping = time.time()
                    continue
                data = json.loads(message['data'])
                if data['task_id'] != task_id:
                    continue
                yield f"data: {json.dumps(data)}\n\n"
                if data['status'] in ('SUCCESS', 'FAILED', 'CANCELED'):
                    break
        finally:
            pubsub.close()
    return StreamingHttpResponse(stream(), content_type='text/event-stream')

三个必须注意的点：

nginx 反代要关掉缓冲，否则 SSE 会被攒着一起发：

nginx location /task/ { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Connection ''; proxy_buffering off; # 关键 proxy_cache off; proxy_read_timeout 3600s; # SSE 连接要长 chunked_transfer_encoding on; }

gunicorn 的 worker 会被长连接占住。SSE 是长连接，同步 worker 模式下每个 SSE 连接占一个 worker。要么用 gevent worker（-k gevent），要么把 SSE 单开服务。我们最后给 SSE 单独跑了一组 gevent worker。
别忘了心跳。中间没数据超过 60 秒，某些代理会直接掐断。我每 15 秒发一个 : ping\n\n（SSE 注释行，客户端会忽略）。

六、重试与幂等：别把 4GB 的文件下两遍

这是我觉得最值得写的一节，因为它直接关系到钱（流量）和时间。

为什么会重复执行

Celery 的 acks_late=True 意味着：任务执行完之后才向 broker 确认。如果 worker 在执行过程中挂了（OOM、机器重启、进程被 kill），消息会被重新投递给别的 worker，任务重新执行。这是设计上的正确行为——保证任务不丢——但如果你没做幂等，后果是"同一个 4GB 文件下了两遍"。

还有一个更隐蔽的原因：visibility_timeout。Redis 作为 broker 时，消息被 worker 取走后会被标记为"不可见"，超过 visibility_timeout（默认 3600 秒）还没确认，Redis 就认为这个 worker 死了，把消息重新放回队列。所以任何可能跑超过 1 小时的任务，必须调大这个值，否则你会看到任务"执行到一半重新开始"。

CELERY_BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 43200}  # 12 小时

我的经验值：visibility_timeout 设成"你预估最长任务耗时 × 2"。

重试怎么写

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(bind=True, autoretry_for=(requests.RequestException,),
             retry_backoff=True, retry_backoff_max=600,
             retry_kwargs={'max_retries': 3},
             soft_time_limit=1800, time_limit=1860)
def download_video(self, task_id):
    ...

autoretry_for：遇到指定异常自动重试，不用手写 self.retry()；
retry_backoff=True：指数退避（1s、2s、4s、8s…），别一秒重试十次把对方服务器打崩；
retry_backoff_max：退避上限；
soft_time_limit：超时抛 SoftTimeLimitExceeded，任务是可以捕获它做清理的；time_limit 是硬超时，直接 kill 进程。两个都配，软的比硬的早 60 秒，给清理留时间。

清理长这样：

@shared_task(bind=True, soft_time_limit=1800, time_limit=1860)
def download_video(self, task_id):
    tmp_path = None
    try:
        tmp_path = start_download(task_id)
        finish(task_id, tmp_path)
    except SoftTimeLimitExceeded:
        # 超时：删半成品，标记失败，别留一地 .part 文件
        if tmp_path and os.path.exists(tmp_path):
            os.remove(tmp_path)
        mark_failed(task_id, 'timeout')
        raise
    except Exception as e:
        mark_failed(task_id, str(e))
        raise

幂等怎么做

三条一起上，我的标准做法：

1. 临时文件名 + 原子改名

tmp_path = final_path + '.part'
download_to(tmp_path)
os.replace(tmp_path, final_path)     # 原子操作，要么没有要么完整

这样即使中途挂了，目录里也只有一个 .part 文件，不会有一个"看起来完整其实是半截"的文件骗过后续流程。

2. 幂等键（Redis 锁）

from django.core.cache import cache

@shared_task(bind=True)
def download_video(self, task_id):
    key = f'dl:lock:{task_id}'
    # set nx=True：只有不存在时才设置成功
    if not cache.add(key, '1', timeout=3600):
        logger.warning('task %s already running, skip', task_id)
        return
    try:
        ...
    finally:
        cache.delete(key)

cache.add() 对应 Redis 的 SETNX，是原子的。注意 timeout 要给足，别任务还在跑锁就过期了。

3. 数据库层面的状态检查

task = DownloadTask.objects.get(id=task_id)
if task.status == 'SUCCESS' and task.file_path and os.path.exists(task.file_path):
    logger.info('already done, skip')
    return

这三条不是多余的：锁防并发，状态检查防重复提交，临时文件防脏数据。缺任何一条，我都在生产上见过对应的事故。

七、并发、预取与资源控制

预取（prefetch）必须关

Celery 默认会预取消息：worker 一次从队列里拿 prefetch_multiplier × concurrency 条消息攒在本地。默认 prefetch_multiplier=4，concurrency=4 就是一个 worker 先拿 16 条消息。

对短任务（毫秒级）这能提升吞吐；对我们这种"一个任务几分钟"的场景是灾难——一个 worker 手里攒着 16 个任务，其他 worker 闲着，任务分配完全不均衡。更要命的是，如果某个任务卡住，它手里那批都得等着。

CELERY_WORKER_PREFETCH_MULTIPLIER = 1

这一行改完，队列的响应时间肉眼可见地变均匀了。

并发数怎么定

看任务类型：

任务类型	瓶颈	建议
解析（网络 IO 为主）	网络	concurrency 可以大（8~16），或者用 gevent/eventlet
下载（网络 + 磁盘）	带宽	按带宽算，别超过 4~8
转码（CPU）	CPU	等于 CPU 核数，甚至更少（留给系统）

CPU 密集型任务不要开超过核数的并发，上下文切换的开销会让总吞吐反而下降。我实测过 8 核机器上转码任务 concurrency=8 和 =16，后者总耗时多了 15%。

按队列拆 worker

# 终端 1：解析队列，IO 密集，并发大
celery -A video_downloader worker -Q parse --concurrency=8 -P gevent

# 终端 2：转码队列，CPU 密集，并发等于核数
celery -A video_downloader worker -Q transcode --concurrency=8 -P prefork

拆开的好处：转码把 CPU 吃满的时候，解析任务依然能秒回。不拆的话，一堆转码任务能把所有 worker 占死，用户点解析就一直排队。

内存泄漏防护

长时间跑的 worker 会有内存缓慢增长（第三方库、循环引用、ffmpeg 子进程残留）。两个参数：

celery -A video_downloader worker --max-tasks-per-child=100 --max-memory-per-child=400000

--max-tasks-per-child=100：每个子进程执行 100 个任务后回收重建；
--max-memory-per-child=400000：单个子进程内存超过 400MB 就回收（单位是 KB）。

我们加上这两个之后，worker 的内存从"三天涨到 3G"变成了稳定在 500MB 左右。代价是任务执行完会有进程重建的开销，对分钟级任务可以忽略。

八、定时任务 beat：重复触发与时区坑

用 django-celery-beat，任务存在数据库里，可以在 admin 后台改，比写死在代码里方便。

INSTALLED_APPS += ['django_celery_beat']

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-files': {
        'task': 'downloader.tasks.cleanup_expired',
        'schedule': crontab(hour=3, minute=0),     # 每天凌晨 3 点
        'options': {'queue': 'default'},
    },
    'retry-failed-tasks': {
        'task': 'downloader.tasks.retry_failed',
        'schedule': 300.0,                          # 每 5 分钟
    },
}

三个坑：

1. beat 只能跑一个实例。 如果你不小心起了两个 beat（比如 supervisor 配了 numprocs=2，或者两台机器都起了），每个任务都会执行两次。我遇到过一次：清理任务跑两遍，第二遍把刚生成的临时文件删了。防护办法是加锁：

@shared_task
def cleanup_expired():
    lock = cache.add('beat:lock:cleanup_expired', '1', timeout=600)
    if not lock:
        logger.info('another instance is running, skip')
        return
    try:
        ...
    finally:
        cache.delete('beat:lock:cleanup_expired')

2. 时区。 Celery 的 crontab 默认用 CELERY_TIMEZONE（没配就是 UTC）。我们的业务时间是北京时间，曾经配了 crontab(hour=3) 以为是凌晨 3 点，实际是 UTC 3 点 = 北京时间 11 点——清理任务在业务高峰跑，把正在下载的文件删了。现在我会显式：

CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = False

并且改完一定去 admin 里看一眼"下次运行时间"对不对。

3. 任务积压。 如果某个周期任务执行时间超过了周期（比如每 5 分钟跑一次，但一次要跑 10 分钟），beat 会继续按时投递，队列越堆越多。周期任务里也要加"上一个还没跑完就跳过"的判断，用上面那个锁就行。

九、Redis 与结果后端的运维

Redis 内存

我们第一次内存告警是 Redis 涨到 8G。查下来主要是两块：结果后端的键没过期、SSE 的 pubsub 订阅没释放。

几个必须做的：

# redis.conf
maxmemory 2gb
maxmemory-policy noeviction     # 当 broker 时千万别用 allkeys-lru

maxmemory-policy 用 noeviction 很重要。如果 Redis 满了开始按 LRU 淘汰 key，而它淘汰的正好是 Celery 队列里的消息，任务就静默消失了。内存满了宁可报错，也别静默丢任务——报错你能发现，丢了发现不了。

然后定期看队列长度：

redis-cli llen celery            # 默认队列
redis-cli llen parse             # 自定义队列

队列长度持续增长说明消费跟不上，要么加 worker，要么查有没有任务卡死。我把这个指标接到了告警里（超过 1000 就发消息）。

结果后端的清理

用 django-celery-results，每次任务都会在 django_celery_results_taskresult 表里插一条。跑几个月之后这个表能有几百万行，查询变慢，还占磁盘。

django-celery-beat 自带一个清理任务，加进 schedule 就行：

CELERY_BEAT_SCHEDULE = {
    'backend-cleanup': {
        'task': 'celery.backend_cleanup',
        'schedule': crontab(hour=4, minute=0),
    },
}

或者手动：

celery -A video_downloader call celery.backend_cleanup

我们设的是保留 7 天（CELERY_RESULT_EXPIRES = 7 * 86400）。

怎么监控

最小可用的三板斧（没上 Prometheus 之前我就是这么干的）：

# 1. worker 活着吗
celery -A video_downloader inspect ping

# 2. 当前活跃任务
celery -A video_downloader inspect active

# 3. 队列长度
redis-cli llen parse

把这三条写进 crontab + 告警脚本，能挡住 80% 的故障。想更直观就装 flower：

celery -A video_downloader flower --port=5555

十、我踩过的坑清单

忘了在 __init__.py 里 import celery app → worker 找不到任务，报 Received unregistered task。
没改 visibility_timeout → 跑 2 小时的任务在第 61 分钟被重新投递，跑了两遍。
prefetch_multiplier 没关 → 任务分配极不均匀，一个 worker 扛 16 个，其他闲着。
只靠 Celery 状态，业务状态没写库 → Redis 重启后所有任务状态丢失，前端一片空白。
revoke() 以为能停掉正在跑的任务 → 停不了，只能让任务自己定期检查取消标记。
重试没做幂等 → 网络抖动重试了一次，4GB 文件下了两遍，带宽翻倍。
update_state 写太频繁 → 800 次 Redis 写入，日志刷屏。
SSE 没关 nginx 缓冲 → 事件被攒着，前端永远显示"解析中"。
SSE 用同步 worker → 几个用户打开页面就把 worker 占满。
起了两个 beat → 定时任务全跑两遍，清理任务删了正在用的文件。
时区没配 → crontab 3 点其实是北京时间 11 点，清理任务在业务高峰跑。
Redis 用了 allkeys-lru → 内存满时淘汰了队列里的消息，任务静默消失。
结果表没清理 → 几百万行，数据库查询变慢，磁盘被吃。
time_limit 和 soft_time_limit 只配了一个 → 硬超时直接 kill，留下满地临时文件。
长任务的临时文件没清理 → /tmp 被 .part 文件塞满，后来加了启动时扫描清理。

最后说点感受。把任务挪出请求这件事，技术上一点都不难——装个 Celery、加个 @shared_task、视图里 delay()，半小时就能跑通。真正花时间的是后面那些：状态怎么表达、进度怎么回传、失败了怎么重试而不重复干活、跑久了怎么不把内存和磁盘撑爆。

这些问题的共同点是：它们在本地开发时全部不会出现。本地你只跑几分钟、只跑一个任务、不会有网络抖动、不会有 OOM。所以我的建议是——上线前先问自己这几个问题，哪怕现在答不上来，至少知道要去查：

这个任务最长可能跑多久？visibility_timeout 够吗？
它被执行两次会怎么样？幂等做了吗？
它失败了，临时文件谁清理？
它跑到一半被 kill，状态会停在什么值？前端会怎么显示？
一个月后，结果表和 Redis 会有多大？

把这五个问题答完，你的异步任务才算真的上线了，而不是"本地能跑"。