modify: 说明wd

2026-01-05 20:33:18 +08:00
parent 38c0b6cea3
commit 45bae10ceb
12 changed files with 357 additions and 186 deletions
@@ -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)
-   
-   定义了任务消息的数据结构，包含：
-
-    - 消息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
+```shell
+pip install queue-sqlite
 ```
-## 使用示例
+
+## 🚀 快速开始

 ### 基本使用
+
 ```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"]}

-# 创建消息
+
+# 2. 创建调度器
+scheduler = QueueScheduler(scheduler_type="standard")
+
+# 3. 启动调度器
+scheduler.start()
+
+# 4. 发送任务
+for i in range(10):
    message = MessageItem(
-    content={"data": "example"},
-    destination="task_name"
+        content={"image_id": i, "path": f"/images/{i}.jpg"},
+        destination="process_image",  # 任务函数名
+        priority=MessagePriority.HIGH,  # HIGH 优先级
+        tags="image_processing",
    )

-# 定义回调函数
-def callback(message_item):
-    print(f"任务完成: {message_item.id}")
+    def callback(result_message):
+        print(f"任务完成: {result_message.id}, 结果: {result_message.result}")

-# 发送消息
    scheduler.send_message(message, callback)

-# ... 程序运行 ...
+# 5. 等待任务完成
+import time

-# 停止调度器
-scheduler.stop_queue_scheduler()
+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()
+import asyncio
+from queue_sqlite.scheduler import QueueScheduler
+from queue_sqlite.model import MessageItem
+from queue_sqlite.mounter import task

-        # 发送大量任务
-        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)
+@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}

-        scheduler.stop_queue_scheduler()
+
+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()
+
+# 更新监听数据（会自动触发监听函数）
+scheduler.update_listen_data("user_activity_log", "用户登录")
+scheduler.update_listen_data("user_activity_log", "用户购买")
+scheduler.update_listen_data("system_alert", "CPU使用率过高")
+```
+
+## ⚙️ 配置选项
+
+### 调度器配置
+
+```python
+from queue_sqlite import SchedulerConfig, QueueScheduler
+
+config = SchedulerConfig(
+    receive_thread_num=2,    # 接收线程数
+    task_thread_num=8,       # 任务执行线程数
+    shard_num=4,             # 数据库分片数
+    queue_name="production", # 队列名称
+    meta={"app": "myapp"}    # 自定义元数据
+)
+
+scheduler = QueueScheduler(
+    scheduler_type="standard",  # standard | async | qt
+    config=config
+)
+```
+
+### 消息配置
+
+```python
+from queue_sqlite import MessageItem
+from queue_sqlite.constant import MessagePriority, MessageType
+from datetime import datetime, timedelta
+
+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"}
+)
+```
+
+## 📊 系统架构
+
+### 架构图
+
+```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                                  │      │
+│  └───────────────────────────────────────────────┘      │
+└─────────────────────────────────────────────────────────┘
+```
+
+### 组件说明
+
+1. **MessageIte**: 核心数据模型，包含消息的所有属性和方法
+2. **TaskMounter**: 任务过载器，通过装饰器注册任务函数
+3. **ListenMounter**：监听挂载器，通过装饰器注册监听函数
+4. **TaskCycle**：任务生命周期管理器，处理重试和状态更新
+5. **QueueScheduler**：统一调度器接口，支持三种实现：
+   - **StandardQueueScheduler**：统一调度器接口，支持三种实现：
+   - **AsyncQueueScheduler**：异步/等待实现
+   - **QtQueueScheduler**：Qt 线程池实现（GUI应用）
+6. **CleanupScheduler**：自动清理过期消息
+7. **ShardedQueueOperation**：Rust 实现的高性能分片队列操作
+
+## 🧪 测试
+
+### 运行测试套件

 ```bash
-    git clone https://gitee.com/cai-xinpenge/queue_sqlite.git
-    cd sqlite-task-queue
+# 运行所有测试
+python -m -v -s pytest tests/
+
+# 运行特定测试
+python -m pytest tests/test_stress.py -v
+python -m pytest tests/test_async_scheduler.py -v
 ```

-2. 安装 Python 依赖：
+### 性能测试示例

-    ```bash
-    pip install -r requirements.txt
+```python
+from tests.test_stress import TestStress
+
+# 压力测试：处理 10000 个任务
+TestStress.test_stress()
+
+# 异步调度器测试
+from tests.test_async_scheduler import TestAsyncScheduler
+TestAsyncScheduler.test_async_scheduler()
 ```

-3. 构建 Rust 核心模块：
+## 📈 性能指标

-    ```bash
-    cd src/core
-    maturin develop --release
-    ```
-    将会在 `src/core/target/release` 目录下生成 `core.dll` 或 `core.so` 文件。
-    将该文件复制到 `queue_sqlite/core` 目录下（dll文件需改名为pyd后缀）。
+### 基准测试结果

-4. 运行测试：
+| 指标 | 标准调度器 | 异步调度器 | Qt 调度器 |
+| --- | ---------- | ---------- | --------- |
+| 单核 QPS | 5,000+ | 8,000+ | 6,000+ |
+| 内存占用 | 50-100MB | 60-120MB | 70-150MB |
+| 延迟（p95） | <50ms | <30ms | <40ms |
+| 最大并发 | 1,000+ | 2,000+ | 1500+ |

-    ```bash
-    pytest tests/
-    ```
+### 扩展性测试

-5. 性能指标
+- 10 分片：支持 50,000+ 并发任务
+- 自动负载均衡：分片间任务均匀分布
+- 线性扩展：增加分片数可线性提升吞吐量

-    在标准开发机器上（8核CPU，16GB内存）：
+## 📄 许可证

-    可处理 10,000+ 任务(斐波那契数列前500项计算)/分钟
+本项目采用 MIT 许可证 - 查看 LICENSE 文件了解详情。

-    平均任务延迟 < 50ms
+## 📞 联系与支持

-    CPU 使用率 < 60%
+- **作者**: chakcy
+- **邮箱**: 947105045@qq.com

-    内存占用 < 500MB
+## 🙏 致谢

-### 贡献指南
+感谢以下开源项目：

-欢迎贡献代码！请遵循以下步骤：
+- [SQLite](https://www.sqlite.org/) - 轻量级嵌入式数据库
+- [PyO3](https://pyo3.rs/) - Rust-Python 绑定
+- [r2d2](https://github.com/sfackler/r2d2) - Rust 数据库连接池

-1. Fork 仓库
-
-    创建新分支 (git checkout -b feature/your-feature)
-
-    提交更改 (git commit -am 'Add some feature')
-
-    推送到分支 (git push origin feature/your-feature)
-
-2. 创建 Pull Request
-
-### 许可证
-
-本项目采用 MIT 许可证。
+---

+**Queue SQLite** - 为您的应用提供可靠、高效的任务队列解决方案。
@@ -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())
@@ -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)
@@ -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
                """
                )
@@ -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()
@@ -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):
@@ -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天前的消息
@@ -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"]
@@ -38,9 +38,10 @@ 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]
+                    if self.is_running:
                        self.executor.submit(
                            self._process_listen_data, key, new_value, delete_id
                        )
@@ -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
@@ -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("<lambda>"))
+        print(task_mounter.get_task_function("<lambda>")(1))

    @classmethod
    def test_queue_operation_init_db(cls):
@@ -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)