modify: pyqt6调度器
This commit is contained in:
@@ -10,6 +10,7 @@
|
||||
|
||||
|
||||
from .message_item import MessageItem
|
||||
from .scheduler_config import SchedulerConfig
|
||||
|
||||
|
||||
__all__ = ["MessageItem"]
|
||||
__all__ = ["MessageItem", "SchedulerConfig"]
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
"""发送消息到队列
|
||||
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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)}")
|
||||
@@ -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 监听调度器已停止")
|
||||
@@ -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)}")
|
||||
@@ -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 接收调度器已停止")
|
||||
@@ -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)}")
|
||||
@@ -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 任务调度器已停止")
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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)}")
|
||||
@@ -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 监听调度器已停止")
|
||||
@@ -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)}")
|
||||
@@ -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 {}
|
||||
@@ -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 接收调度器已停止")
|
||||
@@ -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)}")
|
||||
@@ -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 任务调度器已停止")
|
||||
@@ -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):
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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 调度器
|
||||
)
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user