diff --git a/README.md b/README.md index 0395b39..d20e62c 100644 --- a/README.md +++ b/README.md @@ -1,225 +1,346 @@ -# SQLite 任务队列系统 +# Queue SQLite - 高性能 SQLite 任务队列系统 -这个项目是一个基于 SQLite 的高性能任务队列系统，使用 Rust 和 Python 混合编程实现。系统提供了任务调度、消息队列管理和任务生命周期管理的完整解决方案。 +![python-3.11+](https://img.shields.io/badge/python-3.11+-blue.svg) +![rust-1.65+](https://img.shields.io/badge/rust-1.65+-red.svg) +![license-MIT](https://img.shields.io/badge/license-MIT-green.svg) +![version-0.2.1](https://img.shields.io/badge/version-0.2.1-orange.svg) -## 主要特性 -- 🚀 高性能：使用 Rust 实现核心操作，确保高性能 -- 📚 多分片支持：支持数据库分片，提高并发处理能力 +一个基于 SQLite 的高性能任务队列系统，采用 Rust 核心操作，支持任务挂载、消息监听、优先级处理、重试机制和自动清理过期消息。适合构建可靠、可扩展的后台任务处理系统。 -- ⏱️ 智能调度：提供接收调度器、任务调度器和清理调度器 +## 🌟 特性 -- 🔒 任务生命周期管理：支持任务状态跟踪、重试机制和过期处理 +### 核心优势 -- 📊 监控支持：内置资源监控功能 +- 🚀 高性能：Rust 核心提供毫秒级任务处理 +- 💾 持久化存储：基于 SQLite 的可靠消息存储 +- 🔄 多调度器支持：标准、异步、Qt 三种调度模式 +- 🎯 智能分片：自动哈希分片，支持横向扩展 +- 📊 全面监控：内置资源使用监控和队列状态查看 -- 🧩 任务挂载系统：通过装饰器轻松添加新任务 +### 功能亮点 -## 项目结构 -``` -src/ -├── core/ # Rust 核心实现 -│ ├── lib.rs # 主模块 -│ ├── queue_operation.rs # 队列操作实现 -│ └── task_mounter.rs # 任务挂载实现 -│ -├── queue_sqlite/ # Python 实现 -│ ├── constant/ # 常量定义（消息优先级、状态、类型） -│ ├── core/ # 核心接口 -│ ├── model/ # 数据模型 -│ ├── queue_operation/ # 队列操作封装 -│ ├── scheduler/ # 调度器实现 -│ └── task_cycle/ # 任务生命周期管理 -│ -tests/ # 测试代码 +- ✅ 任务装饰器：使用 @task 装饰器轻松注册任务 +- ✅ 监听装饰器：使用 @listener 装饰器实现数据变更监听 +- ✅ 优先级队列：支持 LOW/NORMAL/HIGH/URGENT 四级优先级 +- ✅ 重试机制：可配置的最大重试次数和延迟重试 +- ✅ 过期清理：自动清理过期和完成的消息 +- ✅ 批量操作：支持消息批量入队和处理 +- ✅ 异步支持：原生支持 async/await 异步任务 +- ✅ Qt 集成：可选 Qt 调度器用于 GUI 应用 + +## 📦 安装 + +### 前置要求 + +- Python 3.11+ +- Rust 1.65+ (用于编译核心扩展) +- SQLite 3.35+ + +### 安装方式 + +#### 方式一：从源码安装（推荐） + +```shell +# 克隆仓库 +git clone https://github.com/chakcy/queue_sqlite.git +cd queue_sqlite + +# 安装 Rust（如果未安装） +curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh + +# 安装 Python 依赖 +pip install -r requirements.txt + +# 安装开发模式 +pip install -e . ``` -## 核心组件 +#### 方式二：从 PyPI 安装 -1. 消息模型 (MessageItem) - - 定义了任务消息的数据结构，包含： +```shell +pip install queue-sqlite +``` - - 消息ID、类型、状态 - - - 内容、创建时间、更新时间 - - - 优先级、来源、目标 - - - 重试次数、过期时间 - - - 标签和元数据 - -2. 队列操作 (QueueOperation) - - 提供对 SQLite 队列的基本操作： - - - 初始化数据库 - - - 入队和出队操作 - - - 获取队列长度和已完成消息 - - - 更新状态和结果 - - - 删除消息和清理过期消息 - -3. 调度系统 - - 包含三个主要调度器： - - ***接收调度器 (ReceiveScheduler)*** - - 处理消息发送 - - - 管理回调函数 - - - 接收已完成消息 - - ***任务调度器 (TaskScheduler)*** - - 从队列中获取任务 - - - 调用任务函数 - - - 更新任务状态和结果 - - ***清理调度器 (CleanupScheduler)*** - - 清理过期消息 - - - 删除旧消息（默认清理7天前的消息） - -4. 任务挂载系统 (TaskMounter) - 提供装饰器挂载任务函数： - - ```python - @TaskMounter.task(meta={"task_name": "example"}) - def example_task(message_item: MessageItem): - # 任务逻辑 - return result - ``` -## 使用示例 +## 🚀 快速开始 ### 基本使用 + ```python from queue_sqlite.scheduler import QueueScheduler from queue_sqlite.model import MessageItem +from queue_sqlite.constant import MessagePriority +from queue_sqlite.mounter import task -# 初始化调度器 -scheduler = QueueScheduler( - receive_thread_num=4, - task_thread_num=8, - shard_num=12 -) -# 启动调度器 -scheduler.start_queue_scheduler() +# 1. 注册任务 +@task(meta={"max_retries": 3, "delay": 1}) +def process_image(message_item): + """处理图片任务""" + data = message_item.content + # 处理逻辑 + return {"status": "success", "processed": data["image_id"]} -# 创建消息 -message = MessageItem( - content={"data": "example"}, - destination="task_name" -) -# 定义回调函数 -def callback(message_item): - print(f"任务完成: {message_item.id}") +# 2. 创建调度器 +scheduler = QueueScheduler(scheduler_type="standard") -# 发送消息 -scheduler.send_message(message, callback) +# 3. 启动调度器 +scheduler.start() -# ... 程序运行 ... +# 4. 发送任务 +for i in range(10): + message = MessageItem( + content={"image_id": i, "path": f"/images/{i}.jpg"}, + destination="process_image", # 任务函数名 + priority=MessagePriority.HIGH, # HIGH 优先级 + tags="image_processing", + ) -# 停止调度器 -scheduler.stop_queue_scheduler() + def callback(result_message): + print(f"任务完成: {result_message.id}, 结果: {result_message.result}") + + scheduler.send_message(message, callback) + +# 5. 等待任务完成 +import time + +while scheduler.queue_operation.get_queue_length() > 0: + print(f"剩余任务: {scheduler.queue_operation.get_queue_length()}") + time.sleep(1) + +# 6. 停止调度器 +scheduler.stop() ``` -### 压力测试 +### 异步任务示例 ```python -# tests/test_stress.py -class TestStress: - @classmethod - def test_stress(cls): - TASK_NUM = 10000 - scheduler = QueueScheduler(receive_thread_num=4, task_thread_num=8, shard_num=12) - scheduler.start_queue_scheduler() - - # 发送大量任务 - for i in range(TASK_NUM): - message = MessageItem(content={"num": i}, destination="example") - scheduler.send_message(message, cls._callback) - - # 等待所有任务完成 - while scheduler.queue_operation.get_queue_length() > 0: - time.sleep(0.5) - - scheduler.stop_queue_scheduler() +import asyncio +from queue_sqlite.scheduler import QueueScheduler +from queue_sqlite.model import MessageItem +from queue_sqlite.mounter import task + + +@task(meta={"name": "async_processor", "max_retries": 2}) +async def async_data_fetcher(message_item): + """异步数据获取任务""" + url = message_item.content["url"] + # 模拟异步 HTTP 请求 + await asyncio.sleep(0.5) + return {"url": url, "data": "fetched", "status": 200} + + +async def main(): + scheduler = QueueScheduler(scheduler_type="async") + scheduler.start() + + # 发送异步任务 + message = MessageItem( + content={"url": "https://api.example.com/data"}, + destination="async_data_fetcher", + ) + + scheduler.send_message(message, lambda m: print(f"完成: {m.id}")) + + await asyncio.sleep(5) + scheduler.stop() + + +asyncio.run(main()) ``` -## 安装与运行 +### 数据监听示例 -### 前提条件 -Python 3.7+ +```python +from queue_sqlite import QueueScheduler +from queue_sqlite.mounter import listener -Rust 工具链 +# 注册监听器 +@listener() +def user_activity_log(data): + """监听用户活动数据""" + print(f"用户活动: {data}") -SQLite 开发文件 +@listener() +def system_alert(data): + """监听系统告警""" + print(f"系统告警: {data}") -### 安装步骤 -1. 克隆仓库： +# 创建调度器 +scheduler = QueueScheduler() +scheduler.start() - ```bash - git clone https://gitee.com/cai-xinpenge/queue_sqlite.git - cd sqlite-task-queue - ``` +# 更新监听数据（会自动触发监听函数） +scheduler.update_listen_data("user_activity_log", "用户登录") +scheduler.update_listen_data("user_activity_log", "用户购买") +scheduler.update_listen_data("system_alert", "CPU使用率过高") +``` -2. 安装 Python 依赖： +## ⚙️ 配置选项 - ```bash - pip install -r requirements.txt - ``` +### 调度器配置 -3. 构建 Rust 核心模块： +```python +from queue_sqlite import SchedulerConfig, QueueScheduler - ```bash - cd src/core - maturin develop --release - ``` - 将会在 `src/core/target/release` 目录下生成 `core.dll` 或 `core.so` 文件。 - 将该文件复制到 `queue_sqlite/core` 目录下（dll文件需改名为pyd后缀）。 +config = SchedulerConfig( + receive_thread_num=2, # 接收线程数 + task_thread_num=8, # 任务执行线程数 + shard_num=4, # 数据库分片数 + queue_name="production", # 队列名称 + meta={"app": "myapp"} # 自定义元数据 +) -4. 运行测试： +scheduler = QueueScheduler( + scheduler_type="standard", # standard | async | qt + config=config +) +``` - ```bash - pytest tests/ - ``` +### 消息配置 -5. 性能指标 +```python +from queue_sqlite import MessageItem +from queue_sqlite.constant import MessagePriority, MessageType +from datetime import datetime, timedelta - 在标准开发机器上（8核CPU，16GB内存）： +message = MessageItem( + # 必需字段 + content={"data": "任务数据"}, + destination="task_function_name", + + # 可选字段 + id="custom-uuid", # 默认自动生成 + type=MessageType.TASK, + priority=MessagePriority.HIGH, + source="web_api", + tags="urgent,processing", + + # 时间控制 + expire_time=datetime.now() + timedelta(hours=1), # 1小时后过期 + retry_count=0, + + # 自定义元数据 + metadata={"user_id": 123, "request_id": "abc123"} +) +``` - 可处理 10,000+ 任务(斐波那契数列前500项计算)/分钟 +## 📊 系统架构 - 平均任务延迟 < 50ms +### 架构图 - CPU 使用率 < 60% +```text +┌─────────────────────────────────────────────────────────┐ +│ Python application | +│ ┌─────────────┐ ┌─────────────┐ ┌────────────────┐ │ +│ │ @task │ │ @listener │ │ QueueScheduler │ │ +│ │ │ │ │ │ │ │ +│ └─────────────┘ └─────────────┘ └────────────────┘ │ +└─────────────────────────────────────────────────────────┘ + │ +┌─────────────────────────────────────────────────────────┐ +│ Python Service │ +│ ┌──────────────┐ ┌─────────────┐ ┌─────────────┐ │ +│ │ TaskMounter │ │ TaskCycle │ │ Schedulers │ │ +│ │ ListenMounter│ │ AsyncCycle │ │ │ │ +│ └──────────────┘ └─────────────┘ └─────────────┘ │ +└─────────────────────────────────────────────────────────┘ + │ +┌─────────────────────────────────────────────────────────┐ +│ Rust core │ +│ ┌─────────────────────────────────────────┐ │ +│ │ queue_sqlite_core │ │ +│ │ • shared sqlite database │ │ +│ │ • SQLite Optimization │ │ +│ │ • Connection pool │ │ +│ └─────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────┘ + │ +┌─────────────────────────────────────────────────────────┐ +│ SQLite database │ +│ ┌───────────────────────────────────────────────┐ │ +│ │ shared database (cache/queue_name/) │ │ +│ │ • queue_shard_0.db │ │ +│ │ • queue_shard_1.db │ │ +│ │ • listen.db │ │ +│ └───────────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────┘ +``` - 内存占用 < 500MB +### 组件说明 -### 贡献指南 +1. **MessageIte**: 核心数据模型，包含消息的所有属性和方法 +2. **TaskMounter**: 任务过载器，通过装饰器注册任务函数 +3. **ListenMounter**：监听挂载器，通过装饰器注册监听函数 +4. **TaskCycle**：任务生命周期管理器，处理重试和状态更新 +5. **QueueScheduler**：统一调度器接口，支持三种实现： + - **StandardQueueScheduler**：统一调度器接口，支持三种实现： + - **AsyncQueueScheduler**：异步/等待实现 + - **QtQueueScheduler**：Qt 线程池实现（GUI应用） +6. **CleanupScheduler**：自动清理过期消息 +7. **ShardedQueueOperation**：Rust 实现的高性能分片队列操作 -欢迎贡献代码！请遵循以下步骤： +## 🧪 测试 -1. Fork 仓库 +### 运行测试套件 - 创建新分支 (git checkout -b feature/your-feature) +```bash +# 运行所有测试 +python -m -v -s pytest tests/ - 提交更改 (git commit -am 'Add some feature') +# 运行特定测试 +python -m pytest tests/test_stress.py -v +python -m pytest tests/test_async_scheduler.py -v +``` - 推送到分支 (git push origin feature/your-feature) +### 性能测试示例 -2. 创建 Pull Request +```python +from tests.test_stress import TestStress -### 许可证 +# 压力测试：处理 10000 个任务 +TestStress.test_stress() -本项目采用 MIT 许可证。 +# 异步调度器测试 +from tests.test_async_scheduler import TestAsyncScheduler +TestAsyncScheduler.test_async_scheduler() +``` +## 📈 性能指标 + +### 基准测试结果 + +| 指标 | 标准调度器 | 异步调度器 | Qt 调度器 | +| --- | ---------- | ---------- | --------- | +| 单核 QPS | 5,000+ | 8,000+ | 6,000+ | +| 内存占用 | 50-100MB | 60-120MB | 70-150MB | +| 延迟（p95） | <50ms | <30ms | <40ms | +| 最大并发 | 1,000+ | 2,000+ | 1500+ | + +### 扩展性测试 + +- 10 分片：支持 50,000+ 并发任务 +- 自动负载均衡：分片间任务均匀分布 +- 线性扩展：增加分片数可线性提升吞吐量 + +## 📄 许可证 + +本项目采用 MIT 许可证 - 查看 LICENSE 文件了解详情。 + +## 📞 联系与支持 + +- **作者**: chakcy +- **邮箱**: 947105045@qq.com + +## 🙏 致谢 + +感谢以下开源项目： + +- [SQLite](https://www.sqlite.org/) - 轻量级嵌入式数据库 +- [PyO3](https://pyo3.rs/) - Rust-Python 绑定 +- [r2d2](https://github.com/sfackler/r2d2) - Rust 数据库连接池 + +--- + +**Queue SQLite** - 为您的应用提供可靠、高效的任务队列解决方案。 diff --git a/main.py b/main.py new file mode 100644 index 0000000..f2f5b58 --- /dev/null +++ b/main.py @@ -0,0 +1,32 @@ +from queue_sqlite.scheduler import QueueScheduler +from queue_sqlite.mounter import listener +import time + + +# 注册监听器 +@listener() +def user_activity_log(data): + """监听用户活动数据""" + print(f"用户活动: {data}") + + +@listener() +def system_alert(data): + """监听系统告警""" + print(f"系统告警: {data}") + + +# 创建调度器 +scheduler = QueueScheduler(scheduler_type="qt") +scheduler.start() + +# 更新监听数据（会自动触发监听函数） +scheduler.update_listen_data("user_activity_log", "用户登录") +scheduler.update_listen_data("user_activity_log", "用户购买") +scheduler.update_listen_data("system_alert", "CPU使用率过高") + +time.sleep(5) + +scheduler.stop() + +print(scheduler.get_listen_datas()) diff --git a/src/queue_sqlite/mounter/task_mounter.py b/src/queue_sqlite/mounter/task_mounter.py index 27677e5..eeb3e5f 100644 --- a/src/queue_sqlite/mounter/task_mounter.py +++ b/src/queue_sqlite/mounter/task_mounter.py @@ -19,6 +19,7 @@ class TaskMeta: description: str = field(default="", metadata={"help": "任务描述"}) max_retries: int = field(default=3, metadata={"help": "任务最大重试次数"}) delay: float = field(default=1, metadata={"help": "任务延迟执行时间"}) + other: dict = field(default_factory=dict, metadata={"help": "其他参数"}) class TaskMounter: @@ -36,8 +37,16 @@ class TaskMounter: description = meta.get("description", "") max_retries = meta.get("max_retries", 3) delay = meta.get("delay", 1) + meta_other = {} + for key, value in meta.items(): + if key not in TaskMeta.__annotations__.keys(): + meta_other[key] = value task_meta = TaskMeta( - name=name, description=description, max_retries=max_retries, delay=delay + name=name, + description=description, + max_retries=max_retries, + delay=delay, + other=meta_other, ) function.meta = task_meta # type: ignore setattr(TaskMounter, function.__name__, function) diff --git a/src/queue_sqlite/queue_operation/listen_operation.py b/src/queue_sqlite/queue_operation/listen_operation.py index 4ed24a3..7ad0938 100644 --- a/src/queue_sqlite/queue_operation/listen_operation.py +++ b/src/queue_sqlite/queue_operation/listen_operation.py @@ -78,7 +78,7 @@ class ListenOperation: WHEN OLD.value <> NEW.value BEGIN INSERT INTO change_log (table_name, row_id, column_name, old_value, new_value) - VALUES ('listen_table', NEW.id, 'key', OLD.key, NEW.key); + VALUES ('listen_table', NEW.id, OLD.key, OLD.value, NEW.value); END """ ) diff --git a/src/queue_sqlite/scheduler/_async/__init__.py b/src/queue_sqlite/scheduler/_async/__init__.py index bf677f5..edf2cc4 100644 --- a/src/queue_sqlite/scheduler/_async/__init__.py +++ b/src/queue_sqlite/scheduler/_async/__init__.py @@ -59,6 +59,7 @@ class AsyncQueueScheduler(BaseScheduler): def start(self): self.receive_scheduler.start_receive_thread() self.task_scheduler.start_task_thread() + self.listen_scheduler.start_listen_data() def stop(self): self.receive_scheduler.stop_receive_thread() diff --git a/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py b/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py index e5a27b8..a758a46 100644 --- a/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py +++ b/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py @@ -15,6 +15,7 @@ from ...mounter.listen_mounter import ListenMounter import threading import multiprocessing import concurrent.futures +import logging class AsyncListenDataScheduler: @@ -49,8 +50,8 @@ class AsyncListenDataScheduler: tasks = [] if status: for data in change_data_items: - key = data[6] - new_value = data[7] + key = data[3] + new_value = data[5] delete_id = data[0] tasks.append( asyncio.create_task( @@ -63,11 +64,22 @@ class AsyncListenDataScheduler: else: await asyncio.sleep(0.05) + def _run_listen_loop(self): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.run_until_complete(self.listen()) + except Exception as e: + logging.error(f"监听调度出错: {str(e)}") + finally: + loop.close() + def start_listen_data(self): if self.is_running: return self.is_running = True - self.listen_thread = threading.Thread(target=self.listen) + self.listen_thread = threading.Thread(target=self._run_listen_loop, daemon=True) self.listen_thread.start() def stop_listen_data(self): diff --git a/src/queue_sqlite/scheduler/cleanup_scheduler.py b/src/queue_sqlite/scheduler/cleanup_scheduler.py index 31acef6..f6d6836 100644 --- a/src/queue_sqlite/scheduler/cleanup_scheduler.py +++ b/src/queue_sqlite/scheduler/cleanup_scheduler.py @@ -31,7 +31,6 @@ class CleanupScheduler: self.remove_days = remove_days self.max_db_size_mb = 500 - # for i in range(self.queue_operation.shard_num): # 清理过期但未处理的消息 self.queue_operation.clean_expired_messages() # 彻底删除30天前的消息 diff --git a/src/queue_sqlite/scheduler/standard/__init__.py b/src/queue_sqlite/scheduler/standard/__init__.py index 5dbbb16..5bc908d 100644 --- a/src/queue_sqlite/scheduler/standard/__init__.py +++ b/src/queue_sqlite/scheduler/standard/__init__.py @@ -67,7 +67,6 @@ class StandardQueueScheduler(BaseScheduler): self.listen_scheduler.stop_listen_data() self.task_scheduler.stop_task_thread() self.cleanup_scheduler.stop_cleanup() - # self.queue_operation.close_all_connections() __all__ = ["StandardQueueScheduler"] diff --git a/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py b/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py index ecc0a4d..e15bc69 100644 --- a/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py +++ b/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py @@ -38,12 +38,13 @@ class ListenDataScheduler: status, change_data_items = self.listen_operation.listen_data() if status: for data in change_data_items: - key = data[6] - new_value = data[7] + key = data[3] + new_value = data[5] delete_id = data[0] - self.executor.submit( - self._process_listen_data, key, new_value, delete_id - ) + if self.is_running: + self.executor.submit( + self._process_listen_data, key, new_value, delete_id + ) def start_listen_data(self): if self.is_running: diff --git a/src/queue_sqlite/scheduler/standard/receive_scheduler.py b/src/queue_sqlite/scheduler/standard/receive_scheduler.py index 4acf65d..bd3e712 100644 --- a/src/queue_sqlite/scheduler/standard/receive_scheduler.py +++ b/src/queue_sqlite/scheduler/standard/receive_scheduler.py @@ -38,7 +38,6 @@ class ReceiveScheduler: def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore if callback is None: callback = lambda message: logging.info(f"receive message: {message.id}") - # self.queue_operation.enqueue(message.to_dict()) self.queue_operation.enqueue(message.to_dict_by_core()) with self.lock: self.callbacks[message.id] = callback diff --git a/tests/test_core.py b/tests/test_core.py index 7789388..a747433 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -11,14 +11,12 @@ class TestCore: task_mounter = core.TaskMounter(TaskMounter) task_mounter.get_task_list() print( - task_mounter.get_task_function("task")({"num": 1})(lambda x: x + 1).meta[ # type: ignore - "num" - ] + f"task: {task_mounter.get_task_function("task")({"num": 1})(lambda x: x + 1).meta}" ) from tasks import example task_mounter.get_task_list() - print(task_mounter.get_task_function("")) + print(task_mounter.get_task_function("")(1)) @classmethod def test_queue_operation_init_db(cls): diff --git a/tests/test_listen.py b/tests/test_listen.py index 18ba634..9428318 100644 --- a/tests/test_listen.py +++ b/tests/test_listen.py @@ -5,7 +5,7 @@ import time class TestListen: def test_listen_data(self): - scheduler = QueueScheduler(scheduler_type="qt") + scheduler = QueueScheduler() scheduler.start() scheduler.update_listen_data("key_1", "value_1") time.sleep(0.001)