modify: qt调度器
This commit is contained in:
@@ -5,13 +5,14 @@
|
||||
@Time : 2025-09-27 17:02:18
|
||||
@Author : chakcy
|
||||
@Email : 947105045@qq.com
|
||||
@description : 监听操作模块
|
||||
@description : 监听操作模块 - 改进版，优化数据库连接
|
||||
"""
|
||||
|
||||
|
||||
from ..mounter.listen_mounter import ListenMounter
|
||||
import sqlite3
|
||||
from typing import List, Tuple, Union
|
||||
import logging
|
||||
|
||||
|
||||
class ListenOperation:
|
||||
@@ -21,118 +22,167 @@ class ListenOperation:
|
||||
self.create_table()
|
||||
|
||||
def _get_connection(self):
|
||||
"""获取数据库连接"""
|
||||
return sqlite3.connect(
|
||||
self.db_dir,
|
||||
check_same_thread=False,
|
||||
"""获取数据库连接 - 改进版"""
|
||||
conn = sqlite3.connect(
|
||||
self.db_dir, check_same_thread=False, timeout=30.0 # 增加超时时间
|
||||
)
|
||||
|
||||
# 设置SQLite优化参数
|
||||
conn.execute("PRAGMA journal_mode=WAL;") # 使用WAL模式提高并发
|
||||
conn.execute("PRAGMA synchronous=NORMAL;")
|
||||
conn.execute("PRAGMA cache_size=-20000;")
|
||||
conn.execute("PRAGMA busy_timeout=5000;") # 设置忙时超时
|
||||
conn.execute("PRAGMA mmap_size=268435456;") # 256MB内存映射
|
||||
|
||||
return conn
|
||||
|
||||
def create_table(self):
|
||||
if len(self.listen_fields) == 0:
|
||||
return
|
||||
conn = self._get_connection()
|
||||
|
||||
# 分别执行每个SQL语句而不是使用executescript
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS listen_table (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
key Text,
|
||||
value JSON
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS 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
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
# 检查触发器是否已存在，如果不存在则创建
|
||||
try:
|
||||
# 分别执行每个SQL语句而不是使用executescript
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TRIGGER IF NOT EXISTS track_value_change
|
||||
AFTER UPDATE OF value ON listen_table
|
||||
FOR EACH ROW
|
||||
WHEN OLD.value <> NEW.value
|
||||
BEGIN
|
||||
INSERT INTO change_log (table_name, row_id, column_name, old_value, new_value)
|
||||
VALUES ('listen_table', NEW.id, 'key', OLD.key, NEW.key);
|
||||
END
|
||||
CREATE TABLE IF NOT EXISTS listen_table (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
key Text,
|
||||
value JSON
|
||||
)
|
||||
"""
|
||||
)
|
||||
except sqlite3.Error:
|
||||
# 如果触发器创建失败，我们继续执行其他操作
|
||||
pass
|
||||
|
||||
conn.execute("PRAGMA journal_mode=WAL;")
|
||||
conn.execute("PRAGMA synchronous=NORMAL;")
|
||||
conn.execute("PRAGMA cache_size=-20000;")
|
||||
conn.execute("PRAGMA mmap_size=1073741824;")
|
||||
conn.execute("PRAGMA temp_store=MEMORY;")
|
||||
conn.execute("PRAGMA busy_timeout=5000;")
|
||||
conn.commit()
|
||||
|
||||
for listen_field in self.listen_fields:
|
||||
sql = """
|
||||
INSERT INTO
|
||||
listen_table (key, value)
|
||||
VALUES
|
||||
(?, ?)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS 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.execute(sql, (listen_field, "null"))
|
||||
)
|
||||
|
||||
# 检查触发器是否已存在，如果不存在则创建
|
||||
try:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TRIGGER IF NOT EXISTS track_value_change
|
||||
AFTER UPDATE OF value ON listen_table
|
||||
FOR EACH ROW
|
||||
WHEN OLD.value <> NEW.value
|
||||
BEGIN
|
||||
INSERT INTO change_log (table_name, row_id, column_name, old_value, new_value)
|
||||
VALUES ('listen_table', NEW.id, 'key', OLD.key, NEW.key);
|
||||
END
|
||||
"""
|
||||
)
|
||||
except sqlite3.Error:
|
||||
# 如果触发器创建失败，我们继续执行其他操作
|
||||
pass
|
||||
|
||||
conn.commit()
|
||||
|
||||
for listen_field in self.listen_fields:
|
||||
# 检查是否已存在该key
|
||||
cursor = conn.execute(
|
||||
"SELECT id FROM listen_table WHERE key = ?", (listen_field,)
|
||||
)
|
||||
if cursor.fetchone() is None:
|
||||
sql = """
|
||||
INSERT INTO
|
||||
listen_table (key, value)
|
||||
VALUES
|
||||
(?, ?)
|
||||
"""
|
||||
conn.execute(sql, (listen_field, "null"))
|
||||
conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"创建监听表失败: {str(e)}")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def listen_data(self) -> Tuple[bool, Union[List[Tuple], str]]:
|
||||
sql = f"""
|
||||
SELECT * FROM change_log where is_delete = 0 ORDER BY id DESC LIMIT 100
|
||||
"""
|
||||
"""获取监听数据 - 改进版"""
|
||||
conn = self._get_connection()
|
||||
result = conn.execute(sql).fetchall()
|
||||
if len(result) == 0:
|
||||
return False, "No data found"
|
||||
return True, result
|
||||
try:
|
||||
sql = """
|
||||
SELECT * FROM change_log where is_delete = 0 ORDER BY id DESC LIMIT 100
|
||||
"""
|
||||
result = conn.execute(sql).fetchall()
|
||||
if len(result) == 0:
|
||||
return False, "No data found"
|
||||
return True, result
|
||||
except Exception as e:
|
||||
logging.error(f"获取监听数据失败: {str(e)}")
|
||||
return False, str(e)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def delete_change_log(self, delete_id):
|
||||
sql = f"""
|
||||
DELETE FROM change_log WHERE id = {delete_id}
|
||||
"""
|
||||
"""删除变更日志 - 改进版"""
|
||||
conn = self._get_connection()
|
||||
conn.execute(sql)
|
||||
try:
|
||||
sql = """
|
||||
DELETE FROM change_log WHERE id = ?
|
||||
"""
|
||||
conn.execute(sql, (delete_id,))
|
||||
conn.commit()
|
||||
except Exception as e:
|
||||
logging.error(f"删除变更日志失败 {delete_id}: {str(e)}")
|
||||
raise e # 重新抛出异常以便上层处理
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def update_listen_data(self, key, value):
|
||||
sql = f"""
|
||||
UPDATE listen_table SET value = '{value}' WHERE key = '{key}'
|
||||
"""
|
||||
"""更新监听数据 - 改进版"""
|
||||
conn = self._get_connection()
|
||||
conn.execute(sql)
|
||||
conn.commit()
|
||||
try:
|
||||
sql = """
|
||||
UPDATE listen_table SET value = ? WHERE key = ?
|
||||
"""
|
||||
conn.execute(sql, (value, key))
|
||||
conn.commit()
|
||||
except Exception as e:
|
||||
logging.error(f"更新监听数据失败 {key}: {str(e)}")
|
||||
raise e
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_value(self, key):
|
||||
sql = f"""
|
||||
SELECT value FROM listen_table WHERE key = '{key}'
|
||||
"""
|
||||
"""获取值 - 改进版"""
|
||||
conn = self._get_connection()
|
||||
result = conn.execute(sql).fetchone()
|
||||
if result is None:
|
||||
try:
|
||||
sql = """
|
||||
SELECT value FROM listen_table WHERE key = ?
|
||||
"""
|
||||
result = conn.execute(sql, (key,)).fetchone()
|
||||
if result is None:
|
||||
return None
|
||||
return result[0]
|
||||
except Exception as e:
|
||||
logging.error(f"获取值失败 {key}: {str(e)}")
|
||||
return None
|
||||
return result[0]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_values(self):
|
||||
sql = f"""
|
||||
SELECT key, value FROM listen_table
|
||||
"""
|
||||
"""获取所有值 - 改进版"""
|
||||
conn = self._get_connection()
|
||||
result = conn.execute(sql).fetchall()
|
||||
return result
|
||||
try:
|
||||
sql = """
|
||||
SELECT key, value FROM listen_table
|
||||
"""
|
||||
result = conn.execute(sql).fetchall()
|
||||
return result
|
||||
except Exception as e:
|
||||
logging.error(f"获取所有值失败: {str(e)}")
|
||||
return []
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
@@ -12,7 +12,7 @@ from .standard import StandardQueueScheduler
|
||||
from ._async import AsyncQueueScheduler
|
||||
from .base import BaseScheduler
|
||||
from ..model import MessageItem, SchedulerConfig
|
||||
from typing import Callable
|
||||
from typing import Callable, Literal
|
||||
import logging
|
||||
|
||||
|
||||
@@ -29,11 +29,13 @@ except ImportError:
|
||||
"Qt is not available, if you want to use it, please install PyQt5/PyQt6/PySide6"
|
||||
)
|
||||
|
||||
SchedulerType = Literal["standard", "async", "qt"]
|
||||
|
||||
|
||||
class QueueScheduler(BaseScheduler):
|
||||
def __init__(
|
||||
self,
|
||||
scheduler_type="standard",
|
||||
scheduler_type: SchedulerType = "standard",
|
||||
config: SchedulerConfig = SchedulerConfig(),
|
||||
):
|
||||
scheduler_class = SCHEDULER_TYPES.get(scheduler_type, None)
|
||||
|
||||
@@ -5,24 +5,31 @@
|
||||
@Time : 2025-10-28 14:22:38
|
||||
@Author : chakcy
|
||||
@Email : 947105045@qq.com
|
||||
@description : 监听调度器
|
||||
@description : 监听调度器 - 改进版，解决数据库锁定问题
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
import sqlite3
|
||||
from .listen_task import QtListenTask
|
||||
from ...queue_operation.listen_operation import ListenOperation
|
||||
from qtpy.QtCore import QThreadPool
|
||||
from qtpy.QtCore import QThreadPool, QMutex, QMutexLocker
|
||||
import threading
|
||||
import time
|
||||
|
||||
|
||||
class QtListenScheduler:
|
||||
"""Qt 监听调度器 - 修复版"""
|
||||
"""Qt 监听调度器 - 改进版，解决数据库锁定问题"""
|
||||
|
||||
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.db_mutex = QMutex() # 数据库操作互斥锁
|
||||
|
||||
# 监听配置
|
||||
self.poll_interval = 0.1 # 轮询间隔（秒）
|
||||
self.max_retries = 5 # 最大重试次数
|
||||
self.retry_delay = 0.2 # 重试延迟（秒）
|
||||
|
||||
# 确保监听表存在
|
||||
self._ensure_listen_tables()
|
||||
@@ -30,68 +37,126 @@ class QtListenScheduler:
|
||||
def _ensure_listen_tables(self):
|
||||
"""确保监听相关的表存在"""
|
||||
try:
|
||||
# 检查表是否存在，如果不存在则创建
|
||||
conn = self.listen_operation._get_connection()
|
||||
# 使用互斥锁保护数据库操作
|
||||
with QMutexLocker(self.db_mutex):
|
||||
# 检查表是否存在，如果不存在则创建
|
||||
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
|
||||
)
|
||||
"""
|
||||
# 检查 change_log 表是否存在
|
||||
cursor = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='change_log'"
|
||||
)
|
||||
conn.commit()
|
||||
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 _safe_listen_data(self):
|
||||
"""安全地获取监听数据，带重试机制"""
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
# 使用互斥锁保护数据库操作
|
||||
with QMutexLocker(self.db_mutex):
|
||||
status, change_data_items = self.listen_operation.listen_data()
|
||||
return status, change_data_items
|
||||
except sqlite3.OperationalError as e:
|
||||
if "database is locked" in str(e) and attempt < self.max_retries - 1:
|
||||
logging.warning(
|
||||
f"数据库锁定，第 {attempt + 1} 次重试获取监听数据..."
|
||||
)
|
||||
time.sleep(self.retry_delay)
|
||||
else:
|
||||
logging.error(f"获取监听数据失败: {str(e)}")
|
||||
return False, []
|
||||
except Exception as e:
|
||||
logging.error(f"获取监听数据时发生未知错误: {str(e)}")
|
||||
return False, []
|
||||
|
||||
return False, []
|
||||
|
||||
def _process_listen_data(self, change_data_items):
|
||||
"""处理监听数据"""
|
||||
processed_count = 0
|
||||
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 = QtListenTask(
|
||||
column_name,
|
||||
new_value,
|
||||
int(delete_id),
|
||||
self.listen_operation,
|
||||
self.db_mutex,
|
||||
self.max_retries,
|
||||
self.retry_delay,
|
||||
)
|
||||
self.thread_pool.start(task) # type: ignore
|
||||
processed_count += 1
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"处理监听数据失败: {str(e)}")
|
||||
|
||||
return processed_count
|
||||
|
||||
def _listen_loop(self):
|
||||
"""监听数据变化循环"""
|
||||
"""监听数据变化循环 - 改进版"""
|
||||
consecutive_errors = 0
|
||||
max_consecutive_errors = 10
|
||||
|
||||
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
|
||||
status, change_data_items = self._safe_listen_data()
|
||||
|
||||
# 使用线程池执行监听任务
|
||||
task = QtListenTask(
|
||||
column_name,
|
||||
new_value,
|
||||
int(delete_id),
|
||||
self.listen_operation,
|
||||
)
|
||||
self.thread_pool.start(task) # type: ignore
|
||||
if status and change_data_items:
|
||||
processed_count = self._process_listen_data(change_data_items)
|
||||
consecutive_errors = 0 # 重置连续错误计数
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"处理监听数据失败: {str(e)}")
|
||||
if processed_count > 0:
|
||||
logging.debug(f"成功处理 {processed_count} 个监听数据项")
|
||||
|
||||
elif not status:
|
||||
# 没有数据或获取失败，短暂休眠
|
||||
time.sleep(self.poll_interval)
|
||||
consecutive_errors += 1
|
||||
|
||||
# 如果连续错误太多，延长休眠时间
|
||||
if consecutive_errors >= max_consecutive_errors:
|
||||
# logging.warning("连续多次获取监听数据失败，延长休眠时间")
|
||||
time.sleep(self.poll_interval * 5)
|
||||
consecutive_errors = 0 # 重置计数
|
||||
else:
|
||||
time.sleep(0.05)
|
||||
# 有状态但没有数据，正常休眠
|
||||
time.sleep(self.poll_interval)
|
||||
consecutive_errors = 0
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"监听循环错误: {str(e)}")
|
||||
time.sleep(0.1)
|
||||
consecutive_errors += 1
|
||||
time.sleep(self.poll_interval * 2) # 出错时延长休眠
|
||||
|
||||
def start(self):
|
||||
"""启动监听调度器"""
|
||||
|
||||
@@ -5,28 +5,40 @@
|
||||
@Time : 2025-10-28 14:30:41
|
||||
@Author : chakcy
|
||||
@Email : 947105045@qq.com
|
||||
@description : 监听任务
|
||||
@description : 监听任务 - 改进版，解决数据库锁定问题
|
||||
"""
|
||||
|
||||
from queue_sqlite.queue_operation.listen_operation import ListenOperation
|
||||
from ...mounter.listen_mounter import ListenMounter
|
||||
import logging
|
||||
from qtpy.QtCore import QRunnable
|
||||
from qtpy.QtCore import QRunnable, QMutexLocker
|
||||
import asyncio
|
||||
from typing import Callable
|
||||
import time
|
||||
import sqlite3
|
||||
|
||||
|
||||
class QtListenTask(QRunnable):
|
||||
"""Qt 监听任务 - 修复版"""
|
||||
"""Qt 监听任务 - 改进版，解决数据库锁定问题"""
|
||||
|
||||
def __init__(
|
||||
self, key: str, value: str, delete_id: int, listen_operation: ListenOperation
|
||||
self,
|
||||
key: str,
|
||||
value: str,
|
||||
delete_id: int,
|
||||
listen_operation: ListenOperation,
|
||||
db_mutex,
|
||||
max_retries=5,
|
||||
retry_delay=0.2,
|
||||
):
|
||||
super().__init__()
|
||||
self.key = key
|
||||
self.value = value
|
||||
self.delete_id = delete_id
|
||||
self.listen_operation = listen_operation
|
||||
self.db_mutex = db_mutex
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
self.setAutoDelete(True)
|
||||
|
||||
def run(self):
|
||||
@@ -43,11 +55,29 @@ class QtListenTask(QRunnable):
|
||||
except Exception as e:
|
||||
logging.error(f"监听函数执行错误 {self.key}: {str(e)}")
|
||||
finally:
|
||||
# 删除变更日志
|
||||
# 安全删除变更日志
|
||||
self._safe_delete_change_log()
|
||||
|
||||
def _safe_delete_change_log(self):
|
||||
"""安全删除变更日志，带重试机制"""
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
self.listen_operation.delete_change_log(self.delete_id)
|
||||
# 使用互斥锁保护数据库操作
|
||||
with QMutexLocker(self.db_mutex):
|
||||
self.listen_operation.delete_change_log(self.delete_id)
|
||||
break # 成功则退出循环
|
||||
except sqlite3.OperationalError as e:
|
||||
if "database is locked" in str(e) and attempt < self.max_retries - 1:
|
||||
logging.debug(
|
||||
f"删除变更日志时数据库锁定，第 {attempt + 1} 次重试..."
|
||||
)
|
||||
time.sleep(self.retry_delay)
|
||||
else:
|
||||
logging.error(f"删除变更日志失败 {self.delete_id}: {str(e)}")
|
||||
break
|
||||
except Exception as e:
|
||||
logging.error(f"删除变更日志失败 {self.delete_id}: {str(e)}")
|
||||
break
|
||||
|
||||
async def _run_async_listener(self, listen_function: Callable):
|
||||
"""执行异步监听函数"""
|
||||
|
||||
Reference in New Issue
Block a user