diff --git a/src/queue_sqlite/model/__init__.py b/src/queue_sqlite/model/__init__.py index 0a18a01..041db3c 100644 --- a/src/queue_sqlite/model/__init__.py +++ b/src/queue_sqlite/model/__init__.py @@ -10,6 +10,7 @@ from .message_item import MessageItem +from .scheduler_config import SchedulerConfig -__all__ = ["MessageItem"] +__all__ = ["MessageItem", "SchedulerConfig"] diff --git a/src/queue_sqlite/model/scheduler_config.py b/src/queue_sqlite/model/scheduler_config.py new file mode 100644 index 0000000..f2537ea --- /dev/null +++ b/src/queue_sqlite/model/scheduler_config.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : scheduler_config.py +@Time : 2025-10-28 10:21:21 +@Author : chakcy +@Email : 947105045@qq.com +@description : 队列调度器配置 +""" + +from dataclasses import dataclass, field +from typing import Dict, Any +import logging + + +@dataclass +class SchedulerConfig: + receive_thread_num: int = field( + default=1, + metadata={"help": "接收消息的线程数"}, + ) + task_thread_num: int = field( + default=4, + metadata={"help": "任务执行线程数"}, + ) + shard_num: int = field( + default=4, + metadata={"help": "分片数"}, + ) + queue_name: str = field( + default="default", + metadata={"help": "队列名称"}, + ) + meta: Dict[str, Any] = field( + default_factory=dict, + metadata={"help": "队列元数据"}, + ) + + @classmethod + def get_field_descriptions(cls) -> Dict[str, str]: + """获取所有字段的描述信息 + + Returns: + Dict[str, str]: 字段名到描述信息的映射 + """ + from dataclasses import fields + + descriptions = {} + for field_info in fields(cls): + if "help" in field_info.metadata: + descriptions[field_info.name] = field_info.metadata["help"] + else: + descriptions[field_info.name] = "无描述信息" + + return descriptions + + @classmethod + def print_field_info(cls): + """打印所有字段的信息，包括类型、默认值和描述""" + from dataclasses import fields + + logging.info(f"=== {cls.__name__} 字段信息 ===") + for field_info in fields(cls): + logging.info( + f"{field_info.name}: {field_info.type} (默认值: {field_info.default})" + ) + logging.info(f" 描述: {field_info.metadata.get('help', '无描述信息')}") + logging.info("-" * 50) diff --git a/src/queue_sqlite/scheduler/__init__.py b/src/queue_sqlite/scheduler/__init__.py index a1c873c..1a6a4e3 100644 --- a/src/queue_sqlite/scheduler/__init__.py +++ b/src/queue_sqlite/scheduler/__init__.py @@ -11,7 +11,7 @@ from .standard import StandardQueueScheduler from ._async import AsyncQueueScheduler from .base import BaseScheduler -from ..model import MessageItem +from ..model import MessageItem, SchedulerConfig from typing import Callable import multiprocessing @@ -27,19 +27,25 @@ try: except ImportError: pass +try: + from .pyqt6 import PyQt6QueueScheduler + + if PyQt6QueueScheduler.is_available(): + SCHEDULER_TYPES["pyqt6"] = PyQt6QueueScheduler +except ImportError: + pass + class QueueScheduler(BaseScheduler): def __init__( self, - receive_thread_num=1, - task_thread_num=multiprocessing.cpu_count() * 2, - shard_num=4, scheduler_type="standard", + config: SchedulerConfig = SchedulerConfig(), ): scheduler_class = SCHEDULER_TYPES.get(scheduler_type, None) if not scheduler_class: raise ValueError(f"Invalid scheduler type: {scheduler_type}") - self.scheduler = scheduler_class(receive_thread_num, task_thread_num, shard_num) + self.scheduler = scheduler_class(config) if self.scheduler: self.queue_operation = self.scheduler.queue_operation diff --git a/src/queue_sqlite/scheduler/_async/__init__.py b/src/queue_sqlite/scheduler/_async/__init__.py index 7749d78..bf677f5 100644 --- a/src/queue_sqlite/scheduler/_async/__init__.py +++ b/src/queue_sqlite/scheduler/_async/__init__.py @@ -19,23 +19,29 @@ from queue_sqlite_core import ShardedQueueOperation import os from ..base import BaseScheduler from ..cleanup_scheduler import CleanupScheduler +from ...model import SchedulerConfig class AsyncQueueScheduler(BaseScheduler): def __init__( - self, receive_thread_num=1, task_thread_num=1, shard_num=4, queue_name="default" + self, + config: SchedulerConfig = SchedulerConfig(), ): - self.queue_operation = ShardedQueueOperation(shard_num, queue_name=queue_name) + self.queue_operation = ShardedQueueOperation( + config.shard_num, queue_name=config.queue_name + ) self.receive_scheduler = AsyncReceiveScheduler( - self.queue_operation, receive_thread_num + self.queue_operation, config.receive_thread_num ) self.listen_operation = ListenOperation( os.path.join(self.queue_operation.db_dir, "listen.db") ) self.listen_operation.create_table() self.listen_scheduler = AsyncListenDataScheduler(self.listen_operation) - self.task_scheduler = AsyncTaskScheduler(self.queue_operation, task_thread_num) + self.task_scheduler = AsyncTaskScheduler( + self.queue_operation, config.task_thread_num + ) self.cleanup_scheduler = CleanupScheduler(self.queue_operation) def send_message(self, message: MessageItem, callback: Callable): diff --git a/src/queue_sqlite/scheduler/base/base_scheduler.py b/src/queue_sqlite/scheduler/base/base_scheduler.py index b6b0231..0d7b690 100644 --- a/src/queue_sqlite/scheduler/base/base_scheduler.py +++ b/src/queue_sqlite/scheduler/base/base_scheduler.py @@ -1,11 +1,14 @@ from abc import ABC, abstractmethod from typing import Callable -from ...model import MessageItem +from ...model import MessageItem, SchedulerConfig class BaseScheduler(ABC): """调度器抽象类""" + def __init__(self, config: SchedulerConfig = SchedulerConfig()): + pass + @abstractmethod def send_message(self, message: MessageItem, callback: Callable): """发送消息到队列 diff --git a/src/queue_sqlite/scheduler/pyqt6/__init__.py b/src/queue_sqlite/scheduler/pyqt6/__init__.py index e69de29..27c9870 100644 --- a/src/queue_sqlite/scheduler/pyqt6/__init__.py +++ b/src/queue_sqlite/scheduler/pyqt6/__init__.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-10-26 10:26:16 +@Author : chakcy +@Email : 947105045@qq.com +@description : pyqt6调度器 - 独立实现 +""" + +import logging +import os +from ..base import BaseScheduler +from ...model import MessageItem, SchedulerConfig +from queue_sqlite_core import ShardedQueueOperation +from ...queue_operation.listen_operation import ListenOperation +from .receive_scheduler import PyQt6ReceiveScheduler +from .listen_scheduler import PyQt6ListenScheduler +from .task_scheduler import PyQt6TaskScheduler +from ..cleanup_scheduler import CleanupScheduler as PyQt6CleanupScheduler +from PyQt6.QtCore import QThreadPool +from PyQt6.QtWidgets import QApplication +from typing import Callable, Dict, Any, List + + +class PyQt6QueueScheduler(BaseScheduler): + """完全独立的 PyQt6 队列调度器 - 修复版""" + + def __init__( + self, + config: SchedulerConfig = SchedulerConfig(), + ): + # 确保有 QApplication 实例（对于 GUI 应用） + self._ensure_qapplication() + + # 初始化队列操作 + self.queue_operation = ShardedQueueOperation( + config.shard_num, config.queue_name + ) + + # 初始化监听操作 + db_dir = f"cache/{config.queue_name}" + os.makedirs(db_dir, exist_ok=True) + self.listen_operation = ListenOperation(f"{db_dir}/listen.db") + + # 确保监听表被创建 + self._ensure_listen_operation_tables() + + # 初始化各个独立的调度器组件 + self.receive_scheduler = PyQt6ReceiveScheduler( + self.queue_operation, config.receive_thread_num + ) + self.task_scheduler = PyQt6TaskScheduler( + self.queue_operation, config.task_thread_num + ) + self.listen_scheduler = PyQt6ListenScheduler(self.listen_operation) + self.cleanup_scheduler = PyQt6CleanupScheduler(self.queue_operation) + + @staticmethod + def is_available() -> bool: + try: + import PyQt6 + + # 检查是否在主线程（GUI 调度器通常要求） + return True + except ImportError: + return False # 依赖库未安装时不可用 + + def _ensure_qapplication(self): + """确保 QApplication 实例存在""" + try: + if not QApplication.instance(): + # 对于非GUI应用，创建无窗口的QApplication + import sys + + self.app = QApplication(sys.argv if hasattr(sys, "argv") else []) + logging.info("已创建 QApplication 实例") + except Exception as e: + logging.warning(f"创建 QApplication 失败: {str(e)}") + + def _ensure_listen_operation_tables(self): + """确保监听操作的表被正确创建""" + try: + # 强制重新创建表 + self.listen_operation.create_table() + + # 额外检查 change_log 表 + conn = self.listen_operation._get_connection() + cursor = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" + ) + if cursor.fetchone() is None: + logging.warning("change_log 表不存在，尝试手动创建") + conn.execute( + """ + CREATE TABLE change_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + table_name TEXT, + row_id INTEGER, + column_name TEXT, + old_value TEXT, + new_value TEXT, + is_delete integer DEFAULT 0, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + conn.commit() + + except Exception as e: + logging.error(f"确保监听表存在失败: {str(e)}") + + def send_message(self, message: MessageItem, callback: Callable): + """发送消息到队列""" + self.receive_scheduler.send_message(message, callback) + + def update_listen_data(self, key: str, value: str): + """更新监听数据""" + self.listen_operation.update_listen_data(key, value) + + def get_listen_datas(self) -> List: + """获取所有监听数据""" + return self.listen_operation.get_values() + + def get_listen_data(self, key: str): + """获取单个监听数据""" + return self.listen_operation.get_value(key) + + def start(self): + """启动所有调度器组件""" + self.receive_scheduler.start() + self.task_scheduler.start() + self.cleanup_scheduler.start_cleanup() + self.listen_scheduler.start() + logging.info("PyQt6 队列调度器已完全启动") + + def stop(self): + """停止所有调度器组件""" + self.listen_scheduler.stop() + self.cleanup_scheduler.stop_cleanup() + self.task_scheduler.stop() + self.receive_scheduler.stop() + logging.info("PyQt6 队列调度器已完全停止") + + def get_queue_info(self) -> Dict[str, Any]: + """获取队列信息""" + try: + return { + "queue_length": self.queue_operation.get_queue_length(), + "shard_num": self.queue_operation.shard_num, + "db_dir": self.queue_operation.db_dir, + "active_threads": QThreadPool.globalInstance().activeThreadCount(), # type: ignore + "max_threads": QThreadPool.globalInstance().maxThreadCount(), # type: ignore + } + except Exception as e: + logging.error(f"获取队列信息失败: {str(e)}") + return {} diff --git a/src/queue_sqlite/scheduler/pyqt6/callback_task.py b/src/queue_sqlite/scheduler/pyqt6/callback_task.py new file mode 100644 index 0000000..3410481 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/callback_task.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : callback_task.py +@Time : 2025-10-28 11:40:02 +@Author : chakcy +@Email : 947105045@qq.com +@description : 回调任务 +""" +import asyncio +import logging +from queue_sqlite_core import ShardedQueueOperation +from ...model import MessageItem +from typing import Callable +from PyQt6.QtCore import QRunnable + + +class PyQt6CallbackTask(QRunnable): + """PyQt6 回调任务""" + + def __init__( + self, + callback: Callable, + message: MessageItem, + queue_operation: ShardedQueueOperation, + ): + super().__init__() + self.callback = callback + self.message = message + self.queue_operation = queue_operation + self.setAutoDelete(True) + + def run(self): + """执行回调函数""" + try: + # 检查是否是协程函数 + if asyncio.iscoroutinefunction(self.callback): + # 对于异步回调，在当前线程中运行事件循环 + asyncio.run(self._run_async_callback()) + else: + self.callback(self.message) + except Exception as e: + logging.error(f"回调执行错误: {str(e)}") + finally: + # 删除消息 + try: + self.queue_operation.delete_message(self.message.id) + except Exception as e: + logging.error(f"删除消息失败 {self.message.id}: {str(e)}") + + async def _run_async_callback(self): + """运行异步回调""" + try: + await self.callback(self.message) + except Exception as e: + logging.error(f"异步回调执行错误: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyqt6/listen_scheduler.py b/src/queue_sqlite/scheduler/pyqt6/listen_scheduler.py new file mode 100644 index 0000000..d2efd17 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/listen_scheduler.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_scheduler.py +@Time : 2025-10-28 14:22:38 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听调度器 +""" +import logging +from .listen_task import PyQt6ListenTask +from ...queue_operation.listen_operation import ListenOperation +from PyQt6.QtCore import QThreadPool +import threading +import time + + +class PyQt6ListenScheduler: + """PyQt6 监听调度器 - 修复版""" + + def __init__(self, listen_operation: ListenOperation): + self.listen_operation = listen_operation + self.is_running = False + self.listen_thread = None + self.thread_pool = QThreadPool.globalInstance() + + # 确保监听表存在 + self._ensure_listen_tables() + + def _ensure_listen_tables(self): + """确保监听相关的表存在""" + try: + # 检查表是否存在，如果不存在则创建 + conn = self.listen_operation._get_connection() + + # 检查 change_log 表是否存在 + cursor = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" + ) + if cursor.fetchone() is None: + logging.info("创建缺失的 change_log 表") + conn.execute( + """ + CREATE TABLE change_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + table_name TEXT, + row_id INTEGER, + column_name TEXT, + old_value TEXT, + new_value TEXT, + is_delete integer DEFAULT 0, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + conn.commit() + + except Exception as e: + logging.error(f"确保监听表存在失败: {str(e)}") + + def _listen_loop(self): + """监听数据变化循环""" + while self.is_running: + try: + status, change_data_items = self.listen_operation.listen_data() + if status: + for data in change_data_items: + try: + # 修正字段索引 + if len(data) >= 8: + delete_id = data[0] # id + table_name = data[1] # table_name + row_id = data[2] # row_id + column_name = data[3] # column_name + old_value = data[4] # old_value + new_value = data[5] # new_value + + # 使用线程池执行监听任务 + task = PyQt6ListenTask( + column_name, + new_value, + int(delete_id), + self.listen_operation, + ) + self.thread_pool.start(task) # type: ignore + + except Exception as e: + logging.error(f"处理监听数据失败: {str(e)}") + else: + time.sleep(0.05) + + except Exception as e: + logging.error(f"监听循环错误: {str(e)}") + time.sleep(0.1) + + def start(self): + """启动监听调度器""" + if self.is_running: + return + + self.is_running = True + self.listen_thread = threading.Thread(target=self._listen_loop, daemon=True) + self.listen_thread.start() + logging.info("PyQt6 监听调度器已启动") + + def stop(self): + """停止监听调度器""" + if not self.is_running: + return + + self.is_running = False + if self.listen_thread and self.listen_thread.is_alive(): + self.listen_thread.join(timeout=3.0) + logging.info("PyQt6 监听调度器已停止") diff --git a/src/queue_sqlite/scheduler/pyqt6/listen_task.py b/src/queue_sqlite/scheduler/pyqt6/listen_task.py new file mode 100644 index 0000000..5043021 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/listen_task.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_task.py +@Time : 2025-10-28 14:30:41 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听任务 +""" + +from queue_sqlite.queue_operation.listen_operation import ListenOperation +from ...mounter.listen_mounter import ListenMounter +import logging +from PyQt6.QtCore import QRunnable +import asyncio +from typing import Callable + + +class PyQt6ListenTask(QRunnable): + """PyQt6 监听任务 - 修复版""" + + def __init__( + self, key: str, value: str, delete_id: int, listen_operation: ListenOperation + ): + super().__init__() + self.key = key + self.value = value + self.delete_id = delete_id + self.listen_operation = listen_operation + self.setAutoDelete(True) + + def run(self): + """执行监听回调""" + try: + listen_function = ListenMounter.get_Listener_function(self.key) + if listen_function: + if asyncio.iscoroutinefunction(listen_function): + # 异步监听函数 + asyncio.run(self._run_async_listener(listen_function)) + else: + # 同步监听函数 + listen_function(self.value) + except Exception as e: + logging.error(f"监听函数执行错误 {self.key}: {str(e)}") + finally: + # 删除变更日志 + try: + self.listen_operation.delete_change_log(self.delete_id) + except Exception as e: + logging.error(f"删除变更日志失败 {self.delete_id}: {str(e)}") + + async def _run_async_listener(self, listen_function: Callable): + """执行异步监听函数""" + try: + await listen_function(self.value) + except Exception as e: + logging.error(f"异步监听函数执行错误 {self.key}: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyqt6/receive_scheduler.py b/src/queue_sqlite/scheduler/pyqt6/receive_scheduler.py new file mode 100644 index 0000000..b2b1914 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/receive_scheduler.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : receive_scheduler.py +@Time : 2025-10-28 14:31:06 +@Author : chakcy +@Email : 947105045@qq.com +@description : 接收调度器 +""" + + +from typing import Callable, Optional +from queue_sqlite_core import ShardedQueueOperation +from PyQt6.QtCore import QThreadPool +import threading +from ...model import MessageItem +import time +import logging +from .callback_task import PyQt6CallbackTask + + +class PyQt6ReceiveScheduler: + """PyQt6 接收调度器 - 独立实现""" + + def __init__( + self, queue_operation: ShardedQueueOperation, receive_thread_num: int = 1 + ): + self.callbacks = {} + self.is_running = False + self.lock = threading.Lock() + self.queue_operation = queue_operation + self.receive_thread = None + self.thread_pool = QThreadPool.globalInstance() + + def send_message(self, message: MessageItem, callback: Optional[Callable] = None): + """发送消息到队列""" + if callback is None: + callback = lambda message: logging.info(f"收到消息: {message.id}") + + # 入队消息 + self.queue_operation.enqueue(message.to_dict_by_core()) + with self.lock: + self.callbacks[message.id] = callback + + def _receive_loop(self): + """接收消息循环""" + while self.is_running: + try: + # 获取已完成的消息 + message_list = self.queue_operation.get_completed_messages() + if message_list: + for message_data in message_list: + try: + message = MessageItem.from_dict(message_data) + + with self.lock: + callback = self.callbacks.pop(message.id, None) + if callback is None: + callback = lambda msg: None + + # 使用线程池执行回调 + task = PyQt6CallbackTask( + callback, message, self.queue_operation + ) + self.thread_pool.start(task) # type: ignore + + except Exception as e: + logging.error(f"处理完成消息失败: {str(e)}") + # 即使处理失败也尝试删除消息 + try: + self.queue_operation.delete_message(message_data["id"]) + except: + pass + else: + time.sleep(0.05) # 短暂休眠避免CPU空转 + + except Exception as e: + logging.error(f"接收消息循环错误: {str(e)}") + time.sleep(0.1) # 出错时稍长休眠 + + def start(self): + """启动接收调度器""" + if self.is_running: + return + + self.is_running = True + self.receive_thread = threading.Thread(target=self._receive_loop, daemon=True) + self.receive_thread.start() + logging.info("PySide6 接收调度器已启动") + + def stop(self): + """停止接收调度器""" + if not self.is_running: + return + + self.is_running = False + if self.receive_thread and self.receive_thread.is_alive(): + self.receive_thread.join(timeout=3.0) + logging.info("PySide6 接收调度器已停止") diff --git a/src/queue_sqlite/scheduler/pyqt6/task_executor.py b/src/queue_sqlite/scheduler/pyqt6/task_executor.py new file mode 100644 index 0000000..52808a3 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/task_executor.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_executor.py +@Time : 2025-10-28 14:31:25 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务执行器 +""" + + +import asyncio +import logging +from typing import Any, Callable, Dict +from datetime import datetime +from ...constant import MessageStatus +from ...mounter.task_mounter import TaskMounter +from queue_sqlite_core import ShardedQueueOperation +from ...model.message_item import MessageItem +from PyQt6.QtCore import QRunnable + + +class PyQt6TaskExecutor(QRunnable): + """PyQt6 任务执行器 - 修复版""" + + def __init__( + self, message_data: Dict[str, Any], queue_operation: ShardedQueueOperation + ): + super().__init__() + self.message_data = message_data + self.queue_operation = queue_operation + self.setAutoDelete(True) + + def run(self): + """执行任务""" + try: + message = MessageItem.from_dict(self.message_data) + + if message.destination == "client": + self._process_client_message(message) + return + + task_function = TaskMounter.get_task_function(message.destination) + if task_function is None: + raise ValueError(f"任务函数未找到: {message.destination}") + + # 执行任务 + if asyncio.iscoroutinefunction(task_function): + # 异步任务 - 在当前线程运行事件循环 + asyncio.run(self._execute_async_task(message, task_function)) + else: + # 同步任务 + self._execute_sync_task(message, task_function) + + except Exception as e: + logging.error(f"任务执行错误: {str(e)}") + # 更新任务状态为失败 + try: + self.queue_operation.update_status( + self.message_data["id"], MessageStatus.FAILED.value + ) + self.queue_operation.update_result( + self.message_data["id"], f'{{"error": "{str(e)}"}}' + ) + except Exception as update_error: + logging.error(f"更新任务状态失败: {str(update_error)}") + + def _process_client_message(self, message: MessageItem): + """处理客户端消息""" + message.status = MessageStatus.COMPLETED + message.result = {"result": "success"} + message.updatetime = datetime.now() + self._update_task_result(message) + + def _execute_sync_task(self, message: MessageItem, task_function: Callable): + """执行同步任务""" + from ...cycle.task_cycle import TaskCycle + + try: + task_cycle = TaskCycle(message, task_function) + task_cycle.run() + + status = task_cycle.get_task_status() + if not status: + raise ValueError("任务未完成") + message.status = status + if message.status == MessageStatus.FAILED: + message.result = {"error": task_cycle.get_task_error()} + else: + # 获取序列化后的结果 + result_str = task_cycle.get_task_result() + try: + import json + + message.result = json.loads(result_str) + except: + message.result = {"result": result_str} + + self._update_task_result(message) + + except Exception as e: + logging.error(f"同步任务执行失败 {message.id}: {str(e)}") + message.status = MessageStatus.FAILED + message.result = {"error": str(e)} + self._update_task_result(message) + + async def _execute_async_task(self, message: MessageItem, task_function: Callable): + """执行异步任务""" + from ...cycle.async_task_cycle import AsyncTaskCycle + + try: + task_cycle = AsyncTaskCycle(message, task_function) + await task_cycle.run() + status = task_cycle.get_task_status() + if not status: + raise ValueError("任务未完成") + message.status = status + if message.status == MessageStatus.FAILED: + message.result = {"error": task_cycle.get_task_error()} + else: + # 获取序列化后的结果 + result_str = task_cycle.get_task_result() + try: + import json + + message.result = json.loads(result_str) + except: + message.result = {"result": result_str} + + self._update_task_result(message) + + except Exception as e: + logging.error(f"异步任务执行失败 {message.id}: {str(e)}") + message.status = MessageStatus.FAILED + message.result = {"error": str(e)} + self._update_task_result(message) + + def _update_task_result(self, message: MessageItem): + """更新任务结果到数据库""" + import json + + try: + # 序列化结果 + result_str = json.dumps(message.result) if message.result else "{}" + self.queue_operation.update_result(message.id, result_str) + self.queue_operation.update_status(message.id, message.status.value) + except Exception as e: + logging.error(f"更新任务结果失败 {message.id}: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyqt6/task_scheduler.py b/src/queue_sqlite/scheduler/pyqt6/task_scheduler.py new file mode 100644 index 0000000..0accd20 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyqt6/task_scheduler.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_scheduler.py +@Time : 2025-10-28 14:19:39 +@Author : chakcy +@Email : 947105045@qq.com +@description : PyQt6 任务调度器 +""" +import logging +from PyQt6.QtCore import QThreadPool +from .task_executor import PyQt6TaskExecutor +from queue_sqlite_core import ShardedQueueOperation +import time +import threading + + +class PyQt6TaskScheduler: + """PyQt6 任务调度器 - 独立实现""" + + def __init__( + self, queue_operation: ShardedQueueOperation, task_thread_num: int = 4 + ): + self.is_running = False + self.queue_operation = queue_operation + self.task_thread = None + self.thread_pool = QThreadPool.globalInstance() + # 设置最大线程数 + self.thread_pool.setMaxThreadCount(max(1, task_thread_num)) # type: ignore + + def _task_loop(self): + """任务处理循环""" + while self.is_running: + try: + # 出队消息进行处理 + message_list = self.queue_operation.dequeue( + size=self.thread_pool.maxThreadCount() * 2 # type: ignore + ) + if message_list: + for message_data in message_list: + # 使用线程池执行任务 + task = PyQt6TaskExecutor(message_data, self.queue_operation) + self.thread_pool.start(task) # type: ignore + else: + time.sleep(0.05) # 短暂休眠 + + except Exception as e: + logging.error(f"任务调度循环错误: {str(e)}") + time.sleep(0.1) + + def start(self): + """启动任务调度器""" + if self.is_running: + return + + self.is_running = True + self.task_thread = threading.Thread(target=self._task_loop, daemon=True) + self.task_thread.start() + logging.info("PySide6 任务调度器已启动") + + def stop(self): + """停止任务调度器""" + if not self.is_running: + return + + self.is_running = False + if self.task_thread and self.task_thread.is_alive(): + self.task_thread.join(timeout=3.0) + logging.info("PySide6 任务调度器已停止") diff --git a/src/queue_sqlite/scheduler/pyside6/__init__.py b/src/queue_sqlite/scheduler/pyside6/__init__.py index 6dc5127..d601686 100644 --- a/src/queue_sqlite/scheduler/pyside6/__init__.py +++ b/src/queue_sqlite/scheduler/pyside6/__init__.py @@ -8,6 +8,150 @@ @description : pyside6调度器 - 独立实现 """ -from .pyside6_scheduler import PySide6QueueScheduler +import logging +import os +from ..base import BaseScheduler +from ...model import MessageItem, SchedulerConfig +from queue_sqlite_core import ShardedQueueOperation +from ...queue_operation.listen_operation import ListenOperation +from .receive_scheduler import PySide6ReceiveScheduler +from .listen_scheduler import PySide6ListenScheduler +from .task_scheduler import PySide6TaskScheduler +from ..cleanup_scheduler import CleanupScheduler as PySide6CleanupScheduler +from PySide6.QtCore import QThreadPool +from PySide6.QtWidgets import QApplication +from typing import Callable, Dict, Any, List -__all__ = ["PySide6QueueScheduler"] + +class PySide6QueueScheduler(BaseScheduler): + """完全独立的 PySide6 队列调度器 - 修复版""" + + def __init__( + self, + config: SchedulerConfig = SchedulerConfig(), + ): + # 确保有 QApplication 实例（对于 GUI 应用） + self._ensure_qapplication() + + # 初始化队列操作 + self.queue_operation = ShardedQueueOperation( + config.shard_num, config.queue_name + ) + + # 初始化监听操作 + db_dir = f"cache/{config.queue_name}" + os.makedirs(db_dir, exist_ok=True) + self.listen_operation = ListenOperation(f"{db_dir}/listen.db") + + # 确保监听表被创建 + self._ensure_listen_operation_tables() + + # 初始化各个独立的调度器组件 + self.receive_scheduler = PySide6ReceiveScheduler( + self.queue_operation, config.receive_thread_num + ) + self.task_scheduler = PySide6TaskScheduler( + self.queue_operation, config.task_thread_num + ) + self.listen_scheduler = PySide6ListenScheduler(self.listen_operation) + self.cleanup_scheduler = PySide6CleanupScheduler(self.queue_operation) + + @staticmethod + def is_available() -> bool: + try: + import PySide6 + + # 检查是否在主线程（GUI 调度器通常要求） + return True + except ImportError: + return False # 依赖库未安装时不可用 + + def _ensure_qapplication(self): + """确保 QApplication 实例存在""" + try: + if not QApplication.instance(): + # 对于非GUI应用，创建无窗口的QApplication + import sys + + self.app = QApplication(sys.argv if hasattr(sys, "argv") else []) + logging.info("已创建 QApplication 实例") + except Exception as e: + logging.warning(f"创建 QApplication 失败: {str(e)}") + + def _ensure_listen_operation_tables(self): + """确保监听操作的表被正确创建""" + try: + # 强制重新创建表 + self.listen_operation.create_table() + + # 额外检查 change_log 表 + conn = self.listen_operation._get_connection() + cursor = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" + ) + if cursor.fetchone() is None: + logging.warning("change_log 表不存在，尝试手动创建") + conn.execute( + """ + CREATE TABLE change_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + table_name TEXT, + row_id INTEGER, + column_name TEXT, + old_value TEXT, + new_value TEXT, + is_delete integer DEFAULT 0, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + conn.commit() + + except Exception as e: + logging.error(f"确保监听表存在失败: {str(e)}") + + def send_message(self, message: MessageItem, callback: Callable): + """发送消息到队列""" + self.receive_scheduler.send_message(message, callback) + + def update_listen_data(self, key: str, value: str): + """更新监听数据""" + self.listen_operation.update_listen_data(key, value) + + def get_listen_datas(self) -> List: + """获取所有监听数据""" + return self.listen_operation.get_values() + + def get_listen_data(self, key: str): + """获取单个监听数据""" + return self.listen_operation.get_value(key) + + def start(self): + """启动所有调度器组件""" + self.receive_scheduler.start() + self.task_scheduler.start() + self.cleanup_scheduler.start_cleanup() + self.listen_scheduler.start() + logging.info("PySide6 队列调度器已完全启动") + + def stop(self): + """停止所有调度器组件""" + self.listen_scheduler.stop() + self.cleanup_scheduler.stop_cleanup() + self.task_scheduler.stop() + self.receive_scheduler.stop() + logging.info("PySide6 队列调度器已完全停止") + + def get_queue_info(self) -> Dict[str, Any]: + """获取队列信息""" + try: + return { + "queue_length": self.queue_operation.get_queue_length(), + "shard_num": self.queue_operation.shard_num, + "db_dir": self.queue_operation.db_dir, + "active_threads": QThreadPool.globalInstance().activeThreadCount(), + "max_threads": QThreadPool.globalInstance().maxThreadCount(), + } + except Exception as e: + logging.error(f"获取队列信息失败: {str(e)}") + return {} diff --git a/src/queue_sqlite/scheduler/pyside6/callback_task.py b/src/queue_sqlite/scheduler/pyside6/callback_task.py new file mode 100644 index 0000000..8a46489 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/callback_task.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : callback_task.py +@Time : 2025-10-28 11:40:02 +@Author : chakcy +@Email : 947105045@qq.com +@description : 回调任务 +""" +import asyncio +import logging +from queue_sqlite_core import ShardedQueueOperation +from ...model import MessageItem +from typing import Callable +from PySide6.QtCore import QRunnable + + +class PySide6CallbackTask(QRunnable): + """PySide6 回调任务""" + + def __init__( + self, + callback: Callable, + message: MessageItem, + queue_operation: ShardedQueueOperation, + ): + super().__init__() + self.callback = callback + self.message = message + self.queue_operation = queue_operation + self.setAutoDelete(True) + + def run(self): + """执行回调函数""" + try: + # 检查是否是协程函数 + if asyncio.iscoroutinefunction(self.callback): + # 对于异步回调，在当前线程中运行事件循环 + asyncio.run(self._run_async_callback()) + else: + self.callback(self.message) + except Exception as e: + logging.error(f"回调执行错误: {str(e)}") + finally: + # 删除消息 + try: + self.queue_operation.delete_message(self.message.id) + except Exception as e: + logging.error(f"删除消息失败 {self.message.id}: {str(e)}") + + async def _run_async_callback(self): + """运行异步回调""" + try: + await self.callback(self.message) + except Exception as e: + logging.error(f"异步回调执行错误: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyside6/listen_scheduler.py b/src/queue_sqlite/scheduler/pyside6/listen_scheduler.py new file mode 100644 index 0000000..2d4c77a --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/listen_scheduler.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_scheduler.py +@Time : 2025-10-28 14:22:38 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听调度器 +""" +import logging +from .listen_task import PySide6ListenTask +from ...queue_operation.listen_operation import ListenOperation +from PySide6.QtCore import QThreadPool +import threading +import time + + +class PySide6ListenScheduler: + """PySide6 监听调度器 - 修复版""" + + def __init__(self, listen_operation: ListenOperation): + self.listen_operation = listen_operation + self.is_running = False + self.listen_thread = None + self.thread_pool = QThreadPool.globalInstance() + + # 确保监听表存在 + self._ensure_listen_tables() + + def _ensure_listen_tables(self): + """确保监听相关的表存在""" + try: + # 检查表是否存在，如果不存在则创建 + conn = self.listen_operation._get_connection() + + # 检查 change_log 表是否存在 + cursor = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" + ) + if cursor.fetchone() is None: + logging.info("创建缺失的 change_log 表") + conn.execute( + """ + CREATE TABLE change_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + table_name TEXT, + row_id INTEGER, + column_name TEXT, + old_value TEXT, + new_value TEXT, + is_delete integer DEFAULT 0, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + conn.commit() + + except Exception as e: + logging.error(f"确保监听表存在失败: {str(e)}") + + def _listen_loop(self): + """监听数据变化循环""" + while self.is_running: + try: + status, change_data_items = self.listen_operation.listen_data() + if status: + for data in change_data_items: + try: + # 修正字段索引 + if len(data) >= 8: + delete_id = data[0] # id + table_name = data[1] # table_name + row_id = data[2] # row_id + column_name = data[3] # column_name + old_value = data[4] # old_value + new_value = data[5] # new_value + + # 使用线程池执行监听任务 + task = PySide6ListenTask( + column_name, + new_value, + int(delete_id), + self.listen_operation, + ) + self.thread_pool.start(task) + + except Exception as e: + logging.error(f"处理监听数据失败: {str(e)}") + else: + time.sleep(0.05) + + except Exception as e: + logging.error(f"监听循环错误: {str(e)}") + time.sleep(0.1) + + def start(self): + """启动监听调度器""" + if self.is_running: + return + + self.is_running = True + self.listen_thread = threading.Thread(target=self._listen_loop, daemon=True) + self.listen_thread.start() + logging.info("PySide6 监听调度器已启动") + + def stop(self): + """停止监听调度器""" + if not self.is_running: + return + + self.is_running = False + if self.listen_thread and self.listen_thread.is_alive(): + self.listen_thread.join(timeout=3.0) + logging.info("PySide6 监听调度器已停止") diff --git a/src/queue_sqlite/scheduler/pyside6/listen_task.py b/src/queue_sqlite/scheduler/pyside6/listen_task.py new file mode 100644 index 0000000..fe27fc9 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/listen_task.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_task.py +@Time : 2025-10-28 14:30:41 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听任务 +""" + +from queue_sqlite.queue_operation.listen_operation import ListenOperation +from ...mounter.listen_mounter import ListenMounter +import logging +from PySide6.QtCore import QRunnable +import asyncio +from typing import Callable + + +class PySide6ListenTask(QRunnable): + """PySide6 监听任务 - 修复版""" + + def __init__( + self, key: str, value: str, delete_id: int, listen_operation: ListenOperation + ): + super().__init__() + self.key = key + self.value = value + self.delete_id = delete_id + self.listen_operation = listen_operation + self.setAutoDelete(True) + + def run(self): + """执行监听回调""" + try: + listen_function = ListenMounter.get_Listener_function(self.key) + if listen_function: + if asyncio.iscoroutinefunction(listen_function): + # 异步监听函数 + asyncio.run(self._run_async_listener(listen_function)) + else: + # 同步监听函数 + listen_function(self.value) + except Exception as e: + logging.error(f"监听函数执行错误 {self.key}: {str(e)}") + finally: + # 删除变更日志 + try: + self.listen_operation.delete_change_log(self.delete_id) + except Exception as e: + logging.error(f"删除变更日志失败 {self.delete_id}: {str(e)}") + + async def _run_async_listener(self, listen_function: Callable): + """执行异步监听函数""" + try: + await listen_function(self.value) + except Exception as e: + logging.error(f"异步监听函数执行错误 {self.key}: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyside6/pyside6_scheduler.py b/src/queue_sqlite/scheduler/pyside6/pyside6_scheduler.py deleted file mode 100644 index e990306..0000000 --- a/src/queue_sqlite/scheduler/pyside6/pyside6_scheduler.py +++ /dev/null @@ -1,669 +0,0 @@ -#!/usr/bin/env python -# -*- encoding: utf-8 -*- -""" -@File : pyside6_scheduler.py -@Time : 2025-10-26 10:30:00 -@Author : chakcy -@Email : 947105045@qq.com -@description : 完全独立的 PySide6 调度器实现 - 修复版 -""" - -from PySide6.QtCore import QThreadPool, QRunnable, QObject, Signal, QTimer -from PySide6.QtWidgets import QApplication -import threading -import time -import logging -import os -import asyncio -from typing import Callable, List, Dict, Any, Optional -from datetime import datetime - -from ...model import MessageItem -from ...constant import MessageStatus, MessagePriority, MessageType -from ...queue_operation.listen_operation import ListenOperation -from ...mounter.listen_mounter import ListenMounter -from ...mounter.task_mounter import TaskMounter -from queue_sqlite_core import ShardedQueueOperation -from ..base import BaseScheduler -from ..cleanup_scheduler import CleanupScheduler - - -class PySide6CallbackTask(QRunnable): - """PySide6 回调任务""" - - def __init__( - self, - callback: Callable, - message: MessageItem, - queue_operation: ShardedQueueOperation, - ): - super().__init__() - self.callback = callback - self.message = message - self.queue_operation = queue_operation - self.setAutoDelete(True) - - def run(self): - """执行回调函数""" - try: - # 检查是否是协程函数 - if asyncio.iscoroutinefunction(self.callback): - # 对于异步回调，在当前线程中运行事件循环 - asyncio.run(self._run_async_callback()) - else: - self.callback(self.message) - except Exception as e: - logging.error(f"回调执行错误: {str(e)}") - finally: - # 删除消息 - try: - self.queue_operation.delete_message(self.message.id) - except Exception as e: - logging.error(f"删除消息失败 {self.message.id}: {str(e)}") - - async def _run_async_callback(self): - """运行异步回调""" - try: - await self.callback(self.message) - except Exception as e: - logging.error(f"异步回调执行错误: {str(e)}") - - -class PySide6TaskExecutor(QRunnable): - """PySide6 任务执行器 - 修复版""" - - def __init__( - self, message_data: Dict[str, Any], queue_operation: ShardedQueueOperation - ): - super().__init__() - self.message_data = message_data - self.queue_operation = queue_operation - self.setAutoDelete(True) - - def run(self): - """执行任务""" - try: - message = MessageItem.from_dict(self.message_data) - - if message.destination == "client": - self._process_client_message(message) - return - - task_function = TaskMounter.get_task_function(message.destination) - if task_function is None: - raise ValueError(f"任务函数未找到: {message.destination}") - - # 执行任务 - if asyncio.iscoroutinefunction(task_function): - # 异步任务 - 在当前线程运行事件循环 - asyncio.run(self._execute_async_task(message, task_function)) - else: - # 同步任务 - self._execute_sync_task(message, task_function) - - except Exception as e: - logging.error(f"任务执行错误: {str(e)}") - # 更新任务状态为失败 - try: - self.queue_operation.update_status( - self.message_data["id"], MessageStatus.FAILED.value - ) - self.queue_operation.update_result( - self.message_data["id"], f'{{"error": "{str(e)}"}}' - ) - except Exception as update_error: - logging.error(f"更新任务状态失败: {str(update_error)}") - - def _process_client_message(self, message: MessageItem): - """处理客户端消息""" - message.status = MessageStatus.COMPLETED - message.result = {"result": "success"} - message.updatetime = datetime.now() - self._update_task_result(message) - - def _execute_sync_task(self, message: MessageItem, task_function: Callable): - """执行同步任务""" - from ...cycle.task_cycle import TaskCycle - - try: - task_cycle = TaskCycle(message, task_function) - task_cycle.run() - - status = task_cycle.get_task_status() - if not status: - raise ValueError("任务未完成") - message.status = status - if message.status == MessageStatus.FAILED: - message.result = {"error": task_cycle.get_task_error()} - else: - # 获取序列化后的结果 - result_str = task_cycle.get_task_result() - try: - import json - - message.result = json.loads(result_str) - except: - message.result = {"result": result_str} - - self._update_task_result(message) - - except Exception as e: - logging.error(f"同步任务执行失败 {message.id}: {str(e)}") - message.status = MessageStatus.FAILED - message.result = {"error": str(e)} - self._update_task_result(message) - - async def _execute_async_task(self, message: MessageItem, task_function: Callable): - """执行异步任务""" - from ...cycle.async_task_cycle import AsyncTaskCycle - - try: - task_cycle = AsyncTaskCycle(message, task_function) - await task_cycle.run() - status = task_cycle.get_task_status() - if not status: - raise ValueError("任务未完成") - message.status = status - if message.status == MessageStatus.FAILED: - message.result = {"error": task_cycle.get_task_error()} - else: - # 获取序列化后的结果 - result_str = task_cycle.get_task_result() - try: - import json - - message.result = json.loads(result_str) - except: - message.result = {"result": result_str} - - self._update_task_result(message) - - except Exception as e: - logging.error(f"异步任务执行失败 {message.id}: {str(e)}") - message.status = MessageStatus.FAILED - message.result = {"error": str(e)} - self._update_task_result(message) - - def _update_task_result(self, message: MessageItem): - """更新任务结果到数据库""" - import json - - try: - # 序列化结果 - result_str = json.dumps(message.result) if message.result else "{}" - self.queue_operation.update_result(message.id, result_str) - self.queue_operation.update_status(message.id, message.status.value) - except Exception as e: - logging.error(f"更新任务结果失败 {message.id}: {str(e)}") - - -class PySide6ListenTask(QRunnable): - """PySide6 监听任务 - 修复版""" - - def __init__( - self, key: str, value: str, delete_id: int, listen_operation: ListenOperation - ): - super().__init__() - self.key = key - self.value = value - self.delete_id = delete_id - self.listen_operation = listen_operation - self.setAutoDelete(True) - - def run(self): - """执行监听回调""" - try: - listen_function = ListenMounter.get_Listener_function(self.key) - if listen_function: - if asyncio.iscoroutinefunction(listen_function): - # 异步监听函数 - asyncio.run(self._run_async_listener(listen_function)) - else: - # 同步监听函数 - listen_function(self.value) - except Exception as e: - logging.error(f"监听函数执行错误 {self.key}: {str(e)}") - finally: - # 删除变更日志 - try: - self.listen_operation.delete_change_log(self.delete_id) - except Exception as e: - logging.error(f"删除变更日志失败 {self.delete_id}: {str(e)}") - - async def _run_async_listener(self, listen_function: Callable): - """执行异步监听函数""" - try: - await listen_function(self.value) - except Exception as e: - logging.error(f"异步监听函数执行错误 {self.key}: {str(e)}") - - -class PySide6ReceiveScheduler: - """PySide6 接收调度器 - 独立实现""" - - def __init__( - self, queue_operation: ShardedQueueOperation, receive_thread_num: int = 1 - ): - self.callbacks = {} - self.is_running = False - self.lock = threading.Lock() - self.queue_operation = queue_operation - self.receive_thread = None - self.thread_pool = QThreadPool.globalInstance() - - def send_message(self, message: MessageItem, callback: Optional[Callable] = None): - """发送消息到队列""" - if callback is None: - callback = lambda message: logging.info(f"收到消息: {message.id}") - - # 入队消息 - self.queue_operation.enqueue(message.to_dict_by_core()) - with self.lock: - self.callbacks[message.id] = callback - - def _receive_loop(self): - """接收消息循环""" - while self.is_running: - try: - # 获取已完成的消息 - message_list = self.queue_operation.get_completed_messages() - if message_list: - for message_data in message_list: - try: - message = MessageItem.from_dict(message_data) - - with self.lock: - callback = self.callbacks.pop(message.id, None) - if callback is None: - callback = lambda msg: None - - # 使用线程池执行回调 - task = PySide6CallbackTask( - callback, message, self.queue_operation - ) - self.thread_pool.start(task) - - except Exception as e: - logging.error(f"处理完成消息失败: {str(e)}") - # 即使处理失败也尝试删除消息 - try: - self.queue_operation.delete_message(message_data["id"]) - except: - pass - else: - time.sleep(0.05) # 短暂休眠避免CPU空转 - - except Exception as e: - logging.error(f"接收消息循环错误: {str(e)}") - time.sleep(0.1) # 出错时稍长休眠 - - def start(self): - """启动接收调度器""" - if self.is_running: - return - - self.is_running = True - self.receive_thread = threading.Thread(target=self._receive_loop, daemon=True) - self.receive_thread.start() - logging.info("PySide6 接收调度器已启动") - - def stop(self): - """停止接收调度器""" - if not self.is_running: - return - - self.is_running = False - if self.receive_thread and self.receive_thread.is_alive(): - self.receive_thread.join(timeout=3.0) - logging.info("PySide6 接收调度器已停止") - - -class PySide6TaskScheduler: - """PySide6 任务调度器 - 独立实现""" - - def __init__( - self, queue_operation: ShardedQueueOperation, task_thread_num: int = 4 - ): - self.is_running = False - self.queue_operation = queue_operation - self.task_thread = None - self.thread_pool = QThreadPool.globalInstance() - # 设置最大线程数 - self.thread_pool.setMaxThreadCount(max(1, task_thread_num)) - - def _task_loop(self): - """任务处理循环""" - while self.is_running: - try: - # 出队消息进行处理 - message_list = self.queue_operation.dequeue( - size=self.thread_pool.maxThreadCount() * 2 - ) - if message_list: - for message_data in message_list: - # 使用线程池执行任务 - task = PySide6TaskExecutor(message_data, self.queue_operation) - self.thread_pool.start(task) - else: - time.sleep(0.05) # 短暂休眠 - - except Exception as e: - logging.error(f"任务调度循环错误: {str(e)}") - time.sleep(0.1) - - def start(self): - """启动任务调度器""" - if self.is_running: - return - - self.is_running = True - self.task_thread = threading.Thread(target=self._task_loop, daemon=True) - self.task_thread.start() - logging.info("PySide6 任务调度器已启动") - - def stop(self): - """停止任务调度器""" - if not self.is_running: - return - - self.is_running = False - if self.task_thread and self.task_thread.is_alive(): - self.task_thread.join(timeout=3.0) - logging.info("PySide6 任务调度器已停止") - - -class PySide6ListenScheduler: - """PySide6 监听调度器 - 修复版""" - - def __init__(self, listen_operation: ListenOperation): - self.listen_operation = listen_operation - self.is_running = False - self.listen_thread = None - self.thread_pool = QThreadPool.globalInstance() - - # 确保监听表存在 - self._ensure_listen_tables() - - def _ensure_listen_tables(self): - """确保监听相关的表存在""" - try: - # 检查表是否存在，如果不存在则创建 - conn = self.listen_operation._get_connection() - - # 检查 change_log 表是否存在 - cursor = conn.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" - ) - if cursor.fetchone() is None: - logging.info("创建缺失的 change_log 表") - conn.execute( - """ - CREATE TABLE change_log ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - table_name TEXT, - row_id INTEGER, - column_name TEXT, - old_value TEXT, - new_value TEXT, - is_delete integer DEFAULT 0, - timestamp DATETIME DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - conn.commit() - - except Exception as e: - logging.error(f"确保监听表存在失败: {str(e)}") - - def _listen_loop(self): - """监听数据变化循环""" - while self.is_running: - try: - status, change_data_items = self.listen_operation.listen_data() - if status: - for data in change_data_items: - try: - # 修正字段索引 - if len(data) >= 8: - delete_id = data[0] # id - table_name = data[1] # table_name - row_id = data[2] # row_id - column_name = data[3] # column_name - old_value = data[4] # old_value - new_value = data[5] # new_value - - # 使用线程池执行监听任务 - task = PySide6ListenTask( - column_name, - new_value, - int(delete_id), - self.listen_operation, - ) - self.thread_pool.start(task) - - except Exception as e: - logging.error(f"处理监听数据失败: {str(e)}") - else: - time.sleep(0.05) - - except Exception as e: - logging.error(f"监听循环错误: {str(e)}") - time.sleep(0.1) - - def start(self): - """启动监听调度器""" - if self.is_running: - return - - self.is_running = True - self.listen_thread = threading.Thread(target=self._listen_loop, daemon=True) - self.listen_thread.start() - logging.info("PySide6 监听调度器已启动") - - def stop(self): - """停止监听调度器""" - if not self.is_running: - return - - self.is_running = False - if self.listen_thread and self.listen_thread.is_alive(): - self.listen_thread.join(timeout=3.0) - logging.info("PySide6 监听调度器已停止") - - -class PySide6CleanupScheduler: - """PySide6 清理调度器 - 独立实现""" - - def __init__( - self, - queue_operation: ShardedQueueOperation, - interval_minutes: int = 60, - remove_days: int = 30, - ): - self.queue_operation = queue_operation - self.interval = interval_minutes * 60 # 转换为秒 - self.remove_days = remove_days - self.is_running = False - self.cleanup_thread = None - - # 立即执行一次清理 - self._perform_cleanup() - - def _cleanup_loop(self): - """清理循环""" - while self.is_running: - try: - self._perform_cleanup() - except Exception as e: - logging.error(f"清理操作错误: {str(e)}") - - # 休眠等待下次清理 - for _ in range(self.interval): - if not self.is_running: - break - time.sleep(1) - - def _perform_cleanup(self): - """执行清理操作""" - try: - # 清理过期但未处理的消息 - self.queue_operation.clean_expired_messages() - # 彻底删除指定天数前的消息 - self.queue_operation.remove_expired_messages(self.remove_days) - except Exception as e: - logging.error(f"执行清理失败: {str(e)}") - - def start(self): - """启动清理调度器""" - if self.is_running: - return - - self.is_running = True - self.cleanup_thread = threading.Thread(target=self._cleanup_loop, daemon=True) - self.cleanup_thread.start() - logging.info("PySide6 清理调度器已启动") - - def stop(self): - """停止清理调度器""" - if not self.is_running: - return - - self.is_running = False - if self.cleanup_thread and self.cleanup_thread.is_alive(): - self.cleanup_thread.join(timeout=3.0) - logging.info("PySide6 清理调度器已停止") - - -class PySide6QueueScheduler(BaseScheduler): - """完全独立的 PySide6 队列调度器 - 修复版""" - - def __init__( - self, - receive_thread_num: int = 1, - task_thread_num: int = 4, - shard_num: int = 4, - queue_name: str = "pyside6_queue", - ): - # 确保有 QApplication 实例（对于 GUI 应用） - self._ensure_qapplication() - - # 初始化队列操作 - self.queue_operation = ShardedQueueOperation(shard_num, queue_name) - - # 初始化监听操作 - db_dir = f"cache/{queue_name}" - os.makedirs(db_dir, exist_ok=True) - self.listen_operation = ListenOperation(f"{db_dir}/listen.db") - - # 确保监听表被创建 - self._ensure_listen_operation_tables() - - # 初始化各个独立的调度器组件 - self.receive_scheduler = PySide6ReceiveScheduler( - self.queue_operation, receive_thread_num - ) - self.task_scheduler = PySide6TaskScheduler( - self.queue_operation, task_thread_num - ) - self.listen_scheduler = PySide6ListenScheduler(self.listen_operation) - self.cleanup_scheduler = PySide6CleanupScheduler(self.queue_operation) - - @staticmethod - def is_available() -> bool: - try: - import PySide6 - - # 检查是否在主线程（GUI 调度器通常要求） - return True - except ImportError: - return False # 依赖库未安装时不可用 - - def _ensure_qapplication(self): - """确保 QApplication 实例存在""" - try: - if not QApplication.instance(): - # 对于非GUI应用，创建无窗口的QApplication - import sys - - self.app = QApplication(sys.argv if hasattr(sys, "argv") else []) - logging.info("已创建 QApplication 实例") - except Exception as e: - logging.warning(f"创建 QApplication 失败: {str(e)}") - - def _ensure_listen_operation_tables(self): - """确保监听操作的表被正确创建""" - try: - # 强制重新创建表 - self.listen_operation.create_table() - - # 额外检查 change_log 表 - conn = self.listen_operation._get_connection() - cursor = conn.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'" - ) - if cursor.fetchone() is None: - logging.warning("change_log 表不存在，尝试手动创建") - conn.execute( - """ - CREATE TABLE change_log ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - table_name TEXT, - row_id INTEGER, - column_name TEXT, - old_value TEXT, - new_value TEXT, - is_delete integer DEFAULT 0, - timestamp DATETIME DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - conn.commit() - - except Exception as e: - logging.error(f"确保监听表存在失败: {str(e)}") - - def send_message(self, message: MessageItem, callback: Callable): - """发送消息到队列""" - self.receive_scheduler.send_message(message, callback) - - def update_listen_data(self, key: str, value: str): - """更新监听数据""" - self.listen_operation.update_listen_data(key, value) - - def get_listen_datas(self) -> List: - """获取所有监听数据""" - return self.listen_operation.get_values() - - def get_listen_data(self, key: str): - """获取单个监听数据""" - return self.listen_operation.get_value(key) - - def start(self): - """启动所有调度器组件""" - self.receive_scheduler.start() - self.task_scheduler.start() - self.cleanup_scheduler.start() - self.listen_scheduler.start() - logging.info("PySide6 队列调度器已完全启动") - - def stop(self): - """停止所有调度器组件""" - self.listen_scheduler.stop() - self.cleanup_scheduler.stop() - self.task_scheduler.stop() - self.receive_scheduler.stop() - logging.info("PySide6 队列调度器已完全停止") - - def get_queue_info(self) -> Dict[str, Any]: - """获取队列信息""" - try: - return { - "queue_length": self.queue_operation.get_queue_length(), - "shard_num": self.queue_operation.shard_num, - "db_dir": self.queue_operation.db_dir, - "active_threads": QThreadPool.globalInstance().activeThreadCount(), - "max_threads": QThreadPool.globalInstance().maxThreadCount(), - } - except Exception as e: - logging.error(f"获取队列信息失败: {str(e)}") - return {} diff --git a/src/queue_sqlite/scheduler/pyside6/receive_scheduler.py b/src/queue_sqlite/scheduler/pyside6/receive_scheduler.py new file mode 100644 index 0000000..3359723 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/receive_scheduler.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : receive_scheduler.py +@Time : 2025-10-28 14:31:06 +@Author : chakcy +@Email : 947105045@qq.com +@description : 接收调度器 +""" + + +from typing import Callable, Optional +from queue_sqlite_core import ShardedQueueOperation +from PySide6.QtCore import QThreadPool +import threading +from ...model import MessageItem +import time +import logging +from .callback_task import PySide6CallbackTask + + +class PySide6ReceiveScheduler: + """PySide6 接收调度器 - 独立实现""" + + def __init__( + self, queue_operation: ShardedQueueOperation, receive_thread_num: int = 1 + ): + self.callbacks = {} + self.is_running = False + self.lock = threading.Lock() + self.queue_operation = queue_operation + self.receive_thread = None + self.thread_pool = QThreadPool.globalInstance() + + def send_message(self, message: MessageItem, callback: Optional[Callable] = None): + """发送消息到队列""" + if callback is None: + callback = lambda message: logging.info(f"收到消息: {message.id}") + + # 入队消息 + self.queue_operation.enqueue(message.to_dict_by_core()) + with self.lock: + self.callbacks[message.id] = callback + + def _receive_loop(self): + """接收消息循环""" + while self.is_running: + try: + # 获取已完成的消息 + message_list = self.queue_operation.get_completed_messages() + if message_list: + for message_data in message_list: + try: + message = MessageItem.from_dict(message_data) + + with self.lock: + callback = self.callbacks.pop(message.id, None) + if callback is None: + callback = lambda msg: None + + # 使用线程池执行回调 + task = PySide6CallbackTask( + callback, message, self.queue_operation + ) + self.thread_pool.start(task) + + except Exception as e: + logging.error(f"处理完成消息失败: {str(e)}") + # 即使处理失败也尝试删除消息 + try: + self.queue_operation.delete_message(message_data["id"]) + except: + pass + else: + time.sleep(0.05) # 短暂休眠避免CPU空转 + + except Exception as e: + logging.error(f"接收消息循环错误: {str(e)}") + time.sleep(0.1) # 出错时稍长休眠 + + def start(self): + """启动接收调度器""" + if self.is_running: + return + + self.is_running = True + self.receive_thread = threading.Thread(target=self._receive_loop, daemon=True) + self.receive_thread.start() + logging.info("PySide6 接收调度器已启动") + + def stop(self): + """停止接收调度器""" + if not self.is_running: + return + + self.is_running = False + if self.receive_thread and self.receive_thread.is_alive(): + self.receive_thread.join(timeout=3.0) + logging.info("PySide6 接收调度器已停止") diff --git a/src/queue_sqlite/scheduler/pyside6/task_executor.py b/src/queue_sqlite/scheduler/pyside6/task_executor.py new file mode 100644 index 0000000..1ef742e --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/task_executor.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_executor.py +@Time : 2025-10-28 14:31:25 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务执行器 +""" + + +import asyncio +import logging +from typing import Any, Callable, Dict +from datetime import datetime +from ...constant import MessageStatus +from ...mounter.task_mounter import TaskMounter +from queue_sqlite_core import ShardedQueueOperation +from ...model.message_item import MessageItem +from PySide6.QtCore import QRunnable + + +class PySide6TaskExecutor(QRunnable): + """PySide6 任务执行器 - 修复版""" + + def __init__( + self, message_data: Dict[str, Any], queue_operation: ShardedQueueOperation + ): + super().__init__() + self.message_data = message_data + self.queue_operation = queue_operation + self.setAutoDelete(True) + + def run(self): + """执行任务""" + try: + message = MessageItem.from_dict(self.message_data) + + if message.destination == "client": + self._process_client_message(message) + return + + task_function = TaskMounter.get_task_function(message.destination) + if task_function is None: + raise ValueError(f"任务函数未找到: {message.destination}") + + # 执行任务 + if asyncio.iscoroutinefunction(task_function): + # 异步任务 - 在当前线程运行事件循环 + asyncio.run(self._execute_async_task(message, task_function)) + else: + # 同步任务 + self._execute_sync_task(message, task_function) + + except Exception as e: + logging.error(f"任务执行错误: {str(e)}") + # 更新任务状态为失败 + try: + self.queue_operation.update_status( + self.message_data["id"], MessageStatus.FAILED.value + ) + self.queue_operation.update_result( + self.message_data["id"], f'{{"error": "{str(e)}"}}' + ) + except Exception as update_error: + logging.error(f"更新任务状态失败: {str(update_error)}") + + def _process_client_message(self, message: MessageItem): + """处理客户端消息""" + message.status = MessageStatus.COMPLETED + message.result = {"result": "success"} + message.updatetime = datetime.now() + self._update_task_result(message) + + def _execute_sync_task(self, message: MessageItem, task_function: Callable): + """执行同步任务""" + from ...cycle.task_cycle import TaskCycle + + try: + task_cycle = TaskCycle(message, task_function) + task_cycle.run() + + status = task_cycle.get_task_status() + if not status: + raise ValueError("任务未完成") + message.status = status + if message.status == MessageStatus.FAILED: + message.result = {"error": task_cycle.get_task_error()} + else: + # 获取序列化后的结果 + result_str = task_cycle.get_task_result() + try: + import json + + message.result = json.loads(result_str) + except: + message.result = {"result": result_str} + + self._update_task_result(message) + + except Exception as e: + logging.error(f"同步任务执行失败 {message.id}: {str(e)}") + message.status = MessageStatus.FAILED + message.result = {"error": str(e)} + self._update_task_result(message) + + async def _execute_async_task(self, message: MessageItem, task_function: Callable): + """执行异步任务""" + from ...cycle.async_task_cycle import AsyncTaskCycle + + try: + task_cycle = AsyncTaskCycle(message, task_function) + await task_cycle.run() + status = task_cycle.get_task_status() + if not status: + raise ValueError("任务未完成") + message.status = status + if message.status == MessageStatus.FAILED: + message.result = {"error": task_cycle.get_task_error()} + else: + # 获取序列化后的结果 + result_str = task_cycle.get_task_result() + try: + import json + + message.result = json.loads(result_str) + except: + message.result = {"result": result_str} + + self._update_task_result(message) + + except Exception as e: + logging.error(f"异步任务执行失败 {message.id}: {str(e)}") + message.status = MessageStatus.FAILED + message.result = {"error": str(e)} + self._update_task_result(message) + + def _update_task_result(self, message: MessageItem): + """更新任务结果到数据库""" + import json + + try: + # 序列化结果 + result_str = json.dumps(message.result) if message.result else "{}" + self.queue_operation.update_result(message.id, result_str) + self.queue_operation.update_status(message.id, message.status.value) + except Exception as e: + logging.error(f"更新任务结果失败 {message.id}: {str(e)}") diff --git a/src/queue_sqlite/scheduler/pyside6/task_scheduler.py b/src/queue_sqlite/scheduler/pyside6/task_scheduler.py new file mode 100644 index 0000000..7e6c683 --- /dev/null +++ b/src/queue_sqlite/scheduler/pyside6/task_scheduler.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_scheduler.py +@Time : 2025-10-28 14:19:39 +@Author : chakcy +@Email : 947105045@qq.com +@description : PySide6 任务调度器 +""" +import logging +from PySide6.QtCore import QThreadPool +from .task_executor import PySide6TaskExecutor +from queue_sqlite_core import ShardedQueueOperation +import time +import threading + + +class PySide6TaskScheduler: + """PySide6 任务调度器 - 独立实现""" + + def __init__( + self, queue_operation: ShardedQueueOperation, task_thread_num: int = 4 + ): + self.is_running = False + self.queue_operation = queue_operation + self.task_thread = None + self.thread_pool = QThreadPool.globalInstance() + # 设置最大线程数 + self.thread_pool.setMaxThreadCount(max(1, task_thread_num)) + + def _task_loop(self): + """任务处理循环""" + while self.is_running: + try: + # 出队消息进行处理 + message_list = self.queue_operation.dequeue( + size=self.thread_pool.maxThreadCount() * 2 + ) + if message_list: + for message_data in message_list: + # 使用线程池执行任务 + task = PySide6TaskExecutor(message_data, self.queue_operation) + self.thread_pool.start(task) + else: + time.sleep(0.05) # 短暂休眠 + + except Exception as e: + logging.error(f"任务调度循环错误: {str(e)}") + time.sleep(0.1) + + def start(self): + """启动任务调度器""" + if self.is_running: + return + + self.is_running = True + self.task_thread = threading.Thread(target=self._task_loop, daemon=True) + self.task_thread.start() + logging.info("PySide6 任务调度器已启动") + + def stop(self): + """停止任务调度器""" + if not self.is_running: + return + + self.is_running = False + if self.task_thread and self.task_thread.is_alive(): + self.task_thread.join(timeout=3.0) + logging.info("PySide6 任务调度器已停止") diff --git a/src/queue_sqlite/scheduler/standard/__init__.py b/src/queue_sqlite/scheduler/standard/__init__.py index 04de86b..5dbbb16 100644 --- a/src/queue_sqlite/scheduler/standard/__init__.py +++ b/src/queue_sqlite/scheduler/standard/__init__.py @@ -19,13 +19,17 @@ from ...queue_operation.listen_operation import ListenOperation from ..cleanup_scheduler import CleanupScheduler import os from ..base import BaseScheduler +from ...model import SchedulerConfig class StandardQueueScheduler(BaseScheduler): def __init__( - self, receive_thread_num=1, task_thread_num=1, shard_num=4, queue_name="default" + self, + config: SchedulerConfig = SchedulerConfig(), ): - self.queue_operation = ShardedQueueOperation(shard_num, queue_name) + self.queue_operation = ShardedQueueOperation( + config.shard_num, config.queue_name + ) self.listen_operation = ListenOperation( os.path.join(self.queue_operation.db_dir, "listen.db") ) @@ -33,9 +37,11 @@ class StandardQueueScheduler(BaseScheduler): self.listen_scheduler = ListenDataScheduler(self.listen_operation) self.receive_scheduler = ReceiveScheduler( - self.queue_operation, receive_thread_num + self.queue_operation, config.receive_thread_num + ) + self.task_scheduler = TaskScheduler( + self.queue_operation, config.task_thread_num ) - self.task_scheduler = TaskScheduler(self.queue_operation, task_thread_num) self.cleanup_scheduler = CleanupScheduler(self.queue_operation) def send_message(self, message: MessageItem, callback: Callable): diff --git a/tests/test_async_scheduler.py b/tests/test_async_scheduler.py index fc2eb65..1e1fe89 100644 --- a/tests/test_async_scheduler.py +++ b/tests/test_async_scheduler.py @@ -45,9 +45,6 @@ class TestAsyncScheduler: monitor_thread = threading.Thread(target=cls.monitor_resources, daemon=True) monitor_thread.start() queue_scheduler = QueueScheduler( - receive_thread_num=4, - task_thread_num=8, - shard_num=12, scheduler_type="async", ) diff --git a/tests/test_pyqt6_scheduler.py b/tests/test_pyqt6_scheduler.py new file mode 100644 index 0000000..ce9eebf --- /dev/null +++ b/tests/test_pyqt6_scheduler.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : test_pyqt6_scheduler.py +@Time : 2025-10-26 10:40:00 +@Author : chakcy +@description : 独立 pyqt6 调度器测试 +""" + +import time +from queue_sqlite.scheduler import QueueScheduler +from queue_sqlite.model import MessageItem +from queue_sqlite.mounter import task + + +@task(meta={"task_name": "pyqt6_example"}) +def pyqt6_example_task(message_item: MessageItem): + """pyqt6 调度器测试任务""" + print(f"处理任务: {message_item.id}") + # 模拟一些工作 + result = sum(i * i for i in range(1000)) + message_item.result = { + "status": "completed", + "result": result, + "task_id": message_item.id, + } + return message_item + + +@task(meta={"task_name": "async_pyqt6_task"}) +async def async_pyqt6_task(message_item: MessageItem): + """异步任务测试""" + import asyncio + + print(f"处理异步任务: {message_item.id}") + await asyncio.sleep(0.1) # 模拟异步操作 + message_item.result = {"status": "async_completed", "task_id": message_item.id} + return message_item + + +def task_callback(message_item: MessageItem): + """任务完成回调""" + print(f"任务完成回调: {message_item.id}, 结果: {message_item.result}") + + +def test_pyqt6_scheduler(): + """测试独立的 pyqt6 调度器""" + + # 创建独立的 pyqt6 调度器 + scheduler = QueueScheduler( + scheduler_type="pyqt6", # 指定使用 pyqt6 调度器 + ) + + print("启动 pyqt6 调度器...") + scheduler.start() + + try: + # 发送同步任务 + for i in range(5): + message = MessageItem( + content={"task_index": i, "type": "sync"}, + destination="pyqt6_example_task", + ) + scheduler.send_message(message, task_callback) + print(f"发送同步消息: {message.id}") + + # 发送异步任务 + for i in range(5): + message = MessageItem( + content={"task_index": i, "type": "async"}, + destination="async_pyqt6_task", + ) + scheduler.send_message(message, task_callback) + print(f"发送异步消息: {message.id}") + + # 等待任务处理 + print("等待任务处理...") + for i in range(10): + queue_info = scheduler.scheduler.get_queue_info() + print(f"队列状态: {queue_info}") + time.sleep(1) + + if queue_info.get("queue_length", 0) == 0: + break + + finally: + print("停止 pyqt6 调度器...") + scheduler.stop() + print("测试完成") + + +if __name__ == "__main__": + test_pyqt6_scheduler() diff --git a/tests/test_pyside6_scheduler.py b/tests/test_pyside6_scheduler.py index 9578b35..722e5a2 100644 --- a/tests/test_pyside6_scheduler.py +++ b/tests/test_pyside6_scheduler.py @@ -48,9 +48,6 @@ def test_pyside6_scheduler(): # 创建独立的 PySide6 调度器 scheduler = QueueScheduler( - receive_thread_num=2, - task_thread_num=6, # 使用 PySide6 线程池 - shard_num=4, scheduler_type="pyside6", # 指定使用 PySide6 调度器 ) diff --git a/tests/test_stress.py b/tests/test_stress.py index 9e1bce4..3721718 100644 --- a/tests/test_stress.py +++ b/tests/test_stress.py @@ -56,9 +56,7 @@ class TestStress: cls.is_monitoring = True monitor_thread = threading.Thread(target=cls.monitor_resources, daemon=True) monitor_thread.start() - queue_scheduler = QueueScheduler( - receive_thread_num=4, task_thread_num=8, shard_num=12 - ) + queue_scheduler = QueueScheduler() # threads = [] # for message_item in messages: