diff --git a/.python-version b/.python-version index 9b1d4ba..24ee5b1 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.13t +3.13 diff --git a/scripts/to_json_utils.py b/scripts/to_json_utils.py deleted file mode 100644 index dd66c48..0000000 --- a/scripts/to_json_utils.py +++ /dev/null @@ -1,224 +0,0 @@ -#!/usr/bin/env python -# -*- encoding: utf-8 -*- -""" -@File : to_json_utils.py -@Time : 2025-09-27 16:36:18 -@Author : chakcy -@Email : 947105045@qq.com -@description : xxxxxxxxx -""" - -import sqlite3 -import json -import tkinter as tk -from tkinter import ttk, filedialog, messagebox -from datetime import datetime - - -class SQLiteToJSONConverter: - def __init__(self, root): - self.root = root - self.root.title("SQLite 表转 JSON 工具") - self.root.geometry("800x600") - - # 变量 - self.db_path = tk.StringVar() - self.tables = [] - self.selected_table = tk.StringVar() - - self.create_widgets() - - def create_widgets(self): - # 主框架 - main_frame = ttk.Frame(self.root, padding="10") - main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # type: ignore - - # 数据库选择部分 - db_frame = ttk.LabelFrame(main_frame, text="数据库选择", padding="5") - db_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) # type: ignore - - ttk.Entry(db_frame, textvariable=self.db_path, width=70).grid( - row=0, column=0, padx=5 - ) - ttk.Button(db_frame, text="浏览", command=self.browse_db).grid( - row=0, column=1, padx=5 - ) - ttk.Button(db_frame, text="连接", command=self.connect_db).grid( - row=0, column=2, padx=5 - ) - - # 表选择部分 - table_frame = ttk.LabelFrame(main_frame, text="表选择", padding="5") - table_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) # type: ignore - - ttk.Label(table_frame, text="选择表:").grid(row=0, column=0, sticky=tk.W) - self.table_combo = ttk.Combobox( - table_frame, textvariable=self.selected_table, state="readonly" - ) - self.table_combo.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5) # type: ignore - ttk.Button(table_frame, text="加载数据", command=self.load_table_data).grid( - row=0, column=2, padx=5 - ) - - # 数据显示部分 - data_frame = ttk.LabelFrame(main_frame, text="表数据", padding="5") - data_frame.grid( - row=2, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=5 # type: ignore - ) - - # 创建Treeview显示数据 - columns = ("#0",) - self.tree = ttk.Treeview(data_frame, columns=columns, show="headings") - vsb = ttk.Scrollbar(data_frame, orient="vertical", command=self.tree.yview) - hsb = ttk.Scrollbar(data_frame, orient="horizontal", command=self.tree.xview) - self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) - - self.tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # type: ignore - vsb.grid(row=0, column=1, sticky=(tk.N, tk.S)) # type: ignore - hsb.grid(row=1, column=0, sticky=(tk.W, tk.E)) # type: ignore - - # 按钮部分 - button_frame = ttk.Frame(main_frame) - button_frame.grid(row=3, column=0, columnspan=2, pady=10) - - ttk.Button(button_frame, text="导出为JSON", command=self.export_to_json).pack( - side=tk.LEFT, padx=5 - ) - ttk.Button(button_frame, text="清空", command=self.clear_all).pack( - side=tk.LEFT, padx=5 - ) - ttk.Button(button_frame, text="退出", command=self.root.quit).pack( - side=tk.LEFT, padx=5 - ) - - # 配置权重 - self.root.columnconfigure(0, weight=1) - self.root.rowconfigure(0, weight=1) - main_frame.columnconfigure(0, weight=1) - main_frame.rowconfigure(2, weight=1) - data_frame.columnconfigure(0, weight=1) - data_frame.rowconfigure(0, weight=1) - - def browse_db(self): - file_path = filedialog.askopenfilename( - title="选择SQLite数据库文件", - filetypes=[ - ("SQLite数据库", "*.db *.sqlite *.sqlite3"), - ("所有文件", "*.*"), - ], - ) - if file_path: - self.db_path.set(file_path) - - def connect_db(self): - if not self.db_path.get(): - messagebox.showerror("错误", "请先选择数据库文件") - return - - try: - conn = sqlite3.connect(self.db_path.get()) - cursor = conn.cursor() - - # 获取所有表名 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") - self.tables = [table[0] for table in cursor.fetchall()] - - self.table_combo["values"] = self.tables - if self.tables: - self.selected_table.set(self.tables[0]) - - conn.close() - messagebox.showinfo( - "成功", f"成功连接到数据库，找到 {len(self.tables)} 个表" - ) - - except sqlite3.Error as e: - messagebox.showerror("数据库错误", f"连接数据库时出错: {str(e)}") - - def load_table_data(self): - if not self.selected_table.get(): - messagebox.showerror("错误", "请先选择表") - return - - try: - conn = sqlite3.connect(self.db_path.get()) - conn.row_factory = sqlite3.Row # 这样可以使用列名访问数据 - cursor = conn.cursor() - - # 获取表数据 - cursor.execute(f"SELECT * FROM {self.selected_table.get()}") - rows = cursor.fetchall() - - # 清空Treeview - for item in self.tree.get_children(): - self.tree.delete(item) - - # 设置列 - columns = [description[0] for description in cursor.description] - self.tree["columns"] = columns - for col in columns: - self.tree.heading(col, text=col) - self.tree.column(col, width=100, minwidth=50) - - # 添加数据 - for row in rows: - self.tree.insert("", "end", values=tuple(row)) - - conn.close() - - except sqlite3.Error as e: - messagebox.showerror("数据库错误", f"加载数据时出错: {str(e)}") - - def export_to_json(self): - if not self.selected_table.get(): - messagebox.showerror("错误", "请先选择表") - return - - # 选择保存位置 - file_path = filedialog.asksaveasfilename( - title="保存JSON文件", - defaultextension=".json", - filetypes=[("JSON文件", "*.json"), ("所有文件", "*.*")], - ) - - if not file_path: - return - - try: - conn = sqlite3.connect(self.db_path.get()) - conn.row_factory = sqlite3.Row # 使用列名访问数据 - cursor = conn.cursor() - - # 获取表数据 - cursor.execute(f"SELECT * FROM {self.selected_table.get()}") - rows = cursor.fetchall() - - # 转换为字典列表 - result = [dict(row) for row in rows] - - # 写入JSON文件 - with open(file_path, "w", encoding="utf-8") as f: - json.dump(result, f, indent=4, ensure_ascii=False) - - conn.close() - messagebox.showinfo("成功", f"数据已成功导出到 {file_path}") - - except sqlite3.Error as e: - messagebox.showerror("数据库错误", f"导出数据时出错: {str(e)}") - except Exception as e: - messagebox.showerror("错误", f"保存文件时出错: {str(e)}") - - def clear_all(self): - self.db_path.set("") - self.tables = [] - self.table_combo["values"] = [] - self.selected_table.set("") - for item in self.tree.get_children(): - self.tree.delete(item) - self.tree["columns"] = ("#0",) - - -if __name__ == "__main__": - root = tk.Tk() - app = SQLiteToJSONConverter(root) - root.mainloop() diff --git a/src/queue_sqlite/queue_operation/listen_operation.py b/src/queue_sqlite/queue_operation/listen_operation.py index 19583b4..4ed24a3 100644 --- a/src/queue_sqlite/queue_operation/listen_operation.py +++ b/src/queue_sqlite/queue_operation/listen_operation.py @@ -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() diff --git a/src/queue_sqlite/scheduler/__init__.py b/src/queue_sqlite/scheduler/__init__.py index d5d4bd8..7dd1726 100644 --- a/src/queue_sqlite/scheduler/__init__.py +++ b/src/queue_sqlite/scheduler/__init__.py @@ -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) diff --git a/src/queue_sqlite/scheduler/qt/listen_scheduler.py b/src/queue_sqlite/scheduler/qt/listen_scheduler.py index 3f4b0f0..28e286e 100644 --- a/src/queue_sqlite/scheduler/qt/listen_scheduler.py +++ b/src/queue_sqlite/scheduler/qt/listen_scheduler.py @@ -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): """启动监听调度器""" diff --git a/src/queue_sqlite/scheduler/qt/listen_task.py b/src/queue_sqlite/scheduler/qt/listen_task.py index dd61a3b..e43d553 100644 --- a/src/queue_sqlite/scheduler/qt/listen_task.py +++ b/src/queue_sqlite/scheduler/qt/listen_task.py @@ -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): """执行异步监听函数""" diff --git a/tests/test_listen.py b/tests/test_listen.py index 9428318..18ba634 100644 --- a/tests/test_listen.py +++ b/tests/test_listen.py @@ -5,7 +5,7 @@ import time class TestListen: def test_listen_data(self): - scheduler = QueueScheduler() + scheduler = QueueScheduler(scheduler_type="qt") scheduler.start() scheduler.update_listen_data("key_1", "value_1") time.sleep(0.001)