简介

Binomic 是一个自定义异步任务框架：以装饰器声明任务，以 Redis Streams 投递消息， 由多进程 Worker 消费执行，Master 进程负责监督与故障回收。消息体采用 msgspec JSON 序列化，任务 ID 使用 UUIDv7。

特性

• 声明式任务：@task() 装饰器，支持 direct（立即）、delay（延迟）、 cron（周期）三种模式
• Redis Streams 消息代理：基于消费者组与 PEL 的可靠投递，支持通过 xautoclaim 回收失联消费者的消息
• 多进程 Worker：Master 以 multiprocessing 启动多个 Worker 子进程，并携 presence 心跳持续监督
• 类型安全：全量 mypy strict 与 ruff 检查，任务注册表基于泛型 TaskSpec[P, T]
• Litestar 集成：内置 BinomicPlugin，将 Binomic 客户端注入 Litestar 依赖

安装

# uv
uv add binomic

# pip
pip install binomic

要求 Python ≥ 3.12，Redis ≥ 7.x。

快速上手

1. 声明任务

在应用包内的 tasks.py 模块中用装饰器声明任务（Worker 通过 autodiscover 自动发现所有 tasks.py 模块）：

# myapp/tasks.py
import time

from binomic.task import task


@task()
def example(index: int = 0) -> None:
    print(f"[{index}] Current time: {time.time()}")


@task(mode="delay", delay=10)  # 延迟 10 秒执行
def delayed() -> None: ...


@task(mode="cron", cron="*/5 * * * *")  # 每 5 分钟执行
def periodic() -> None: ...

2. 启动并投递

import time

from binomic.client import Binomic, BinomicFactory
from binomic.config import BinomicConfig
from binomic.message import Message
from binomic.task import autodiscover

# 发现 myapp 包内所有 tasks.py 模块并注册任务
autodiscover("myapp")

factory = BinomicFactory(
    broker_dsn="redis://localhost:6379/0",
    redis_dsn="redis://localhost:6379/1",
    module_name="myapp",
    config=BinomicConfig(queues=["default"], workers=2, concurrency=5),
)

binomic: Binomic = factory.create()

# 进入上下文后启动 Master（拉起 worker 子进程）并提供提交入口
async with binomic:
    await binomic.submit(
        Message(name="example", queue="default", enqueued_at=time.time())
    )

3. Litestar 应用中集成

from binomic.config import BinomicConfig
from binomic.plugin.litestar import BinomicPlugin
from litestar import Litestar

app = Litestar(
    plugins=[
        BinomicPlugin(
            app_name="myapp",
            broker_dsn="redis://localhost:6379/0",
            redis_dsn="redis://localhost:6379/1",
            config=BinomicConfig(queues=["default"]),
        )
    ],
)

处理器内通过 binomic 参数注入 Binomic 客户端。

架构

flowchart LR
    P[生产者<br>submit] --> B[AsyncredisBroker<br>Redis Streams]
    B -->|consumer group / PEL| W1[Worker 0<br>子进程]
    B -->|consumer group / PEL| W2[Worker 1<br>子进程]
    M[Master<br>监督进程] --> W1
    M --> W2
    M -->|presence 心跳| R[(Redis)]

• Producer：Binomic.submit 经 BrokerFactory 按 broker DSN 协议创建代理并 enqueue 消息。
• Broker：AsyncredisBroker 将消息写入 Redis Streams，Worker 侧以消费者组 读取；失联消费者的 PEL 消息由 reclaim（xautoclaim）回收重投。
• Master / Worker：Master 以 multiprocessing 拉起 Worker 子进程并监督其存活； Worker 在进程内通过 anyio 以配置并发执行任务，执行结果经 ack 确认。

开发

make install    # 初始化虚拟环境并安装依赖
make check      # lint + 类型检查 + 全部测试
make coverage   # 覆盖率报告（fail_under = 80）
make changelog  # 由 git-cliff 再生成 CHANGELOG.md

提交信息遵循 Conventional Commits， 详见 CONTRIBUTING.md。

变更日志

CHANGELOG.md 由 git-cliff 从提交历史生成， 禁止手动编辑。

许可证

MIT