Backends API

chronflow 提供多种队列后端实现，支持不同的使用场景。

QueueBackend (抽象基类)

QueueBackend

Bases: ABC

队列后端的抽象基类。

connect() abstractmethod async

建立与后端的连接。

disconnect() abstractmethod async

关闭与后端的连接。

enqueue(task_id, task_name, scheduled_time, payload, priority=0) abstractmethod async

将任务加入队列。

参数：

名称	类型	描述	默认
task_id	str	任务的唯一标识	必需
task_name	str	任务名称	必需
scheduled_time	datetime	任务计划执行时间	必需
payload	dict[str, Any]	任务数据和参数	必需
priority	int	任务优先级(数值越大越优先)	0

dequeue(limit=1) abstractmethod async

从队列中获取就绪任务。

参数：

名称	类型	描述	默认
limit	int	最多获取的任务数量	1

返回：

类型	描述
list[dict[str, Any]]	已准备执行的任务负载列表

acknowledge(task_id) abstractmethod async

标记任务成功完成。

参数：

名称	类型	描述	默认
task_id	str	已完成任务的标识	必需

reject(task_id, requeue=False) abstractmethod async

标记任务执行失败。

参数：

名称	类型	描述	默认
task_id	str	失败任务的标识	必需
requeue	bool	是否将任务重新入队	False

get_queue_size() abstractmethod async

获取队列中的待处理任务数量。

clear() abstractmethod async

清空队列中的所有任务。

health_check() abstractmethod async

检查后端是否健康且可访问。

MemoryBackend

MemoryBackend(max_size=10000)

Bases: QueueBackend

基于 heapq 的内存优先队列实现。

特点: - 默认后端,无需任何外部依赖 - 适用于单进程应用和测试场景

初始化内存后端。

参数：

名称	类型	描述	默认
max_size	int	队列允许的最大任务数量	10000

connect() async

内存后端无需建立连接。

disconnect() async

内存后端无需显式断开连接。

enqueue(task_id, task_name, scheduled_time, payload, priority=0) async

将任务加入优先队列。

dequeue(limit=1) async

从队列中获取已就绪的任务。

acknowledge(task_id) async

将任务从待处理集合中移除。

reject(task_id, requeue=False) async

处理失败任务。

get_queue_size() async

获取队列中的任务总数。

clear() async

清除所有任务。

health_check() async

内存后端始终处于健康状态。

__repr__()

返回可读的信息。

SQLiteBackend

SQLiteBackend(db_path='chronflow.db', table_name='task_queue')

Bases: QueueBackend

基于 SQLite 的本地持久化队列实现。

特性: - 本地文件持久化,无需外部服务 - 支持任务优先级和调度时间 - 适合单机部署场景 - 轻量级,零配置

适用场景: - 开发环境和测试 - 单机应用 - 需要持久化但不需要分布式的场景

初始化 SQLite 后端。

参数：

名称	类型	描述	默认
db_path	str | Path	数据库文件路径	'chronflow.db'
table_name	str	任务队列表名	'task_queue'

connect() async

连接到 SQLite 数据库并创建表。

disconnect() async

关闭数据库连接。

enqueue(task_id, task_name, scheduled_time, payload, priority=0) async

将任务添加到队列。

dequeue(limit=1) async

从队列获取就绪的任务。

acknowledge(task_id) async

确认任务成功完成并从队列删除。

reject(task_id, requeue=False) async

拒绝任务(失败处理)。

get_queue_size() async

获取队列中的任务数量。

clear() async

清空队列中的所有任务。

health_check() async

检查数据库健康状态。

get_failed_tasks(limit=100) async

获取失败的任务列表。

参数：

名称	类型	描述	默认
limit	int	返回的最大任务数	100

返回：

类型	描述
list[dict[str, Any]]	失败任务列表

cleanup_old_tasks(days=7) async

清理旧任务记录。

参数：

名称	类型	描述	默认
days	int	保留最近多少天的任务	7

返回：

类型	描述
int	删除的任务数量

__repr__()

字符串表示。

RedisBackend

RedisBackend(url='redis://localhost:6379/0', queue_name='chronflow:queue', pending_set='chronflow:pending', max_connections=10)

Bases: QueueBackend

适用于分布式系统的 Redis 队列实现。

特性: - 分布式任务队列 - 重启后的持久化能力 - 借助 Redis 的高性能 - 支持多工作协程并发

安装: pip install redis

初始化 Redis 后端。

参数：

名称	类型	描述	默认
url	str	Redis 连接 URL	'redis://localhost:6379/0'
queue_name	str	Redis 有序集合的队列名称	'chronflow:queue'
pending_set	str	Redis 集合用于记录待处理任务的名称	'chronflow:pending'
max_connections	int	最大 Redis 连接数	10

connect() async

连接到 Redis。

disconnect() async

断开与 Redis 的连接。

enqueue(task_id, task_name, scheduled_time, payload, priority=0) async

将任务加入 Redis 有序集合。

dequeue(limit=1) async

从 Redis 获取就绪任务。

acknowledge(task_id) async

从待处理集合移除任务。

reject(task_id, requeue=False) async

处理失败任务。

get_queue_size() async

获取队列中的任务总数。

clear() async

清空所有任务。

health_check() async

检查 Redis 的健康状态。

__repr__()

字符串表示。

RabbitMQBackend

RabbitMQBackend(url='amqp://guest:guest@localhost:5672/', queue_name='chronflow_tasks', durable=True, prefetch_count=10)

Bases: QueueBackend

基于 RabbitMQ 的队列实现,适用于高可靠性分布式系统。

特性: - 消息持久化,防止数据丢失 - 支持多个消费者并发处理 - 自动重连机制 - 延迟消息支持(需要 rabbitmq_delayed_message_exchange 插件)

安装: pip install aio-pika

初始化 RabbitMQ 后端。

参数：

名称	类型	描述	默认
url	str	RabbitMQ 连接 URL	'amqp://guest:guest@localhost:5672/'
queue_name	str	队列名称	'chronflow_tasks'
durable	bool	是否持久化队列	True
prefetch_count	int	预取消息数量	10

connect() async

连接到 RabbitMQ。

disconnect() async

断开 RabbitMQ 连接。

enqueue(task_id, task_name, scheduled_time, payload, priority=0) async

将任务添加到 RabbitMQ 队列。

dequeue(limit=1) async

从 RabbitMQ 获取就绪的任务。

acknowledge(task_id) async

确认任务成功完成。

reject(task_id, requeue=False) async

拒绝任务(失败处理)。

get_queue_size() async

获取队列中的任务数量。

clear() async

清空队列中的所有任务。

health_check() async

检查 RabbitMQ 健康状态。

__repr__()

字符串表示。