通过 Redis streams 实现快速、异步、完全类型化的分布式 Python 任务队列

比 arq 快达 14 倍 完全类型化 完善的文档 支持延迟/定时任务 Cron 定时任务 任务中间件 死信队列 流水线 优先队列 支持同步任务（在独立线程中运行） 生产环境支持 Redis Sentinel 与 Cluster 内置 Web UI 用于监控任务 基于 anyio 以结构化并发构建，同时支持 asyncio 和 trio 提示 受够了 redis-py？来看看 coredis，一个快速、完全类型化且支持 Trio 的 Redis 客户端！ $ pip install streaq 快速上手 首先，你需要创建一个 Worker 对象： from streaq import Worker worker = Worker(redis_url="redis://localhost:6379", anyio_backend="trio") 然后像这样向 worker 注册异步任务： import trio async def sleeper(time: int) -> int: await trio.sleep(time) return time async def cronjob() -> None: print("Nobody respects the spammish repetition!") 最后，使用 worker 的异步上下文管理器来入队一些任务： async with worker: await sleeper.enqueue(3)

enqueue 返回一个任务对象，可用于获取结果/信息

task = await sleeper.enqueue(1).start(delay=3) print(await task.info()) print(await task.result(timeout=5)) 把这一切组合起来就得到了 example.py。让我们启动一个 worker： $ streaq run example:worker 然后像这样入队一些任务： $ python example.py 看看输出是什么样的： [INFO] 2025-09-23 02:14:30: starting worker 3265311d for 2 functions [INFO] 2025-09-23 02:14:35: task sleeper □ cf0c55387a214320bd23e8987283a562 → worker 3265311d [INFO] 2025-09-23 02:14:38: task sleeper ■ cf0c55387a214320bd23e8987283a562 ← 3 [INFO] 2025-09-23 02:14:40: task sleeper □ 1de3f192ee4a40d4884ebf303874681c → worker 3265311d [INFO] 2025-09-23 02:14:41: task sleeper ■ 1de3f192ee4a40d4884ebf303874681c ← 1 [INFO] 2025-09-23 02:15:00: task cronjob □ 2a4b864e5ecd4fc99979a92f5db3a6e0 → worker 3265311d Nobody respects the spammish repetition! [INFO] 2025-09-23 02:15:00: task cronjob ■ 2a4b864e5ecd4fc99979a92f5db3a6e0 ← None TaskInfo(fn_name='sleeper', enqueue_time=1751508876961, tries=0, scheduled=datetime.datetime(2025, 7, 3, 2, 14, 39, 961000, tzinfo=datetime.timezone.utc), child=None) TaskResult(fn_name='sleeper', enqueue_time=1751508876961, success=True, start_time=1751508880500, finish_time=1751508881503, tries=1, worker_id='ca5bd9eb', _result=1) 更多示例请查阅文档。