Python 异步任务队列选型：从 Celery 迁到 Arq 的一次实践

背景

项目早期用 Celery + Redis 跑异步任务，功能没问题，但运维成本一直在涨：worker 内存常驻 200MB 起、beat 调度的时区坑、还有任务超时后不重试导致消息丢失。业务量不大（日均 3 万任务），属于典型的"用重锤敲钉子"。后来换成 Arq，代码量砍了一半。

Celery 的三个痛点

**1. 启动开销大**：单个 worker 冷启动要 1.5 秒，容器水平扩容时抖动明显。

**2. 配置分散**：broker、result backend、队列路由、序列化、时区，散落在五六个地方，新人基本看不懂。

**3. 超时语义模糊**：`task_time_limit` 触发后任务直接被杀，队列里的消息状态不一致，需要额外写补偿逻辑。

Arq 的写法

Arq 是基于 asyncio 的轻量队列，一个 worker 就是一个事件循环：

import asyncio
from arq import create_pool
from arq.connections import RedisSettings

async def send_mail(ctx, addr, subject):
    await asyncio.sleep(0.1)
    return f"sent to {addr}"

class WorkerSettings:
    functions = [send_mail]
    redis_settings = RedisSettings(host="127.0.0.1", port=6379)
    max_jobs = 20
    job_timeout = 60

投递任务：

redis = await create_pool(RedisSettings())
await redis.enqueue_job("send_mail", "a@b.com", "hello")

就这么多。没有额外的配置文件，没有 beat 进程（定时用 cron 表达式挂在 WorkerSettings 里）。

迁移过程

• 第一步：把 Celery task 改成 `async def` + 第一个参数 `ctx`，逻辑几乎不用动

• 第二步：`delay()` 换成 `enqueue_job()`，注意参数要可序列化

• 第三步：重试逻辑用 `max_tries` 参数，Arq 自动按指数退避重排

• 第四步：原来的 beat 定时任务迁到 `cron_jobs`，时区显式写 UTC

结果

• worker 内存 200MB → 45MB

• 冷启动 1.5s → 0.2s

• 代码行数 800 → 380

• 任务积压监控沿用 Redis 队列长度，不用改

什么时候不该换

如果任务有复杂的依赖链（chain、chord、group），或者需要跨语言 worker，Celery 生态还是更成熟。Arq 适合"就是想把函数丢到后台跑"的场景，别硬套。

小结

选型的第一原则是匹配规模。日任务不到十万级的时候，队列本身的复杂度往往比业务还高，轻一点反而更稳。