From 149855b8678b34f8d173225b83257bb24234ff6d Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 11 Aug 2025 16:55:55 +0800 Subject: [PATCH] =?UTF-8?q?add:=20=E6=95=B0=E6=8D=AE=E7=9B=91=E5=90=AC?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/queue_sqlite/constant/message_type.py | 5 +- src/queue_sqlite/core/__init__.pyi | 2 +- src/queue_sqlite/cycle/__init__.py | 0 src/queue_sqlite/cycle/listen_cycle.py | 0 .../{task_cycle => cycle}/task_cycle.py | 0 .../{task_cycle => mounter}/__init__.py | 4 +- src/queue_sqlite/mounter/base_mounter.py | 42 +++++++++++ src/queue_sqlite/mounter/listen_mounter.py | 41 +++++++++++ .../{task_cycle => mounter}/task_mounter.py | 0 .../queue_operation/listen_operation.py | 71 +++++++++++++++++++ .../queue_operation/queue_operation.py | 1 - src/queue_sqlite/scheduler/__init__.py | 12 ++++ .../scheduler/listen_data_scheduler.py | 46 ++++++++++++ src/queue_sqlite/scheduler/task_scheduler.py | 4 +- tests/listen/__init__.py | 3 + tests/listen/listen_data.py | 5 ++ tests/tasks/example.py | 2 +- tests/test_core.py | 2 +- tests/test_listen.py | 19 +++++ 19 files changed, 247 insertions(+), 12 deletions(-) create mode 100644 src/queue_sqlite/cycle/__init__.py create mode 100644 src/queue_sqlite/cycle/listen_cycle.py rename src/queue_sqlite/{task_cycle => cycle}/task_cycle.py (100%) rename src/queue_sqlite/{task_cycle => mounter}/__init__.py (55%) create mode 100644 src/queue_sqlite/mounter/base_mounter.py create mode 100644 src/queue_sqlite/mounter/listen_mounter.py rename src/queue_sqlite/{task_cycle => mounter}/task_mounter.py (100%) create mode 100644 src/queue_sqlite/queue_operation/listen_operation.py create mode 100644 src/queue_sqlite/scheduler/listen_data_scheduler.py create mode 100644 tests/listen/__init__.py create mode 100644 tests/listen/listen_data.py create mode 100644 tests/test_listen.py diff --git a/src/queue_sqlite/constant/message_type.py b/src/queue_sqlite/constant/message_type.py index 323ad41..293acfe 100644 --- a/src/queue_sqlite/constant/message_type.py +++ b/src/queue_sqlite/constant/message_type.py @@ -2,6 +2,5 @@ from enum import Enum class MessageType(Enum): - NOTIFICATION = "notification" - SYSTEM = "system" - TASK = "task" + TASK = "task" # 任务 + LISTEN = "listen" # 监听 diff --git a/src/queue_sqlite/core/__init__.pyi b/src/queue_sqlite/core/__init__.pyi index 209b35c..0bea59b 100644 --- a/src/queue_sqlite/core/__init__.pyi +++ b/src/queue_sqlite/core/__init__.pyi @@ -1,4 +1,4 @@ -from ..task_cycle import TaskMounter +from ..mounter import TaskMounter from typing import Callable, List from ..constant import MessageStatus diff --git a/src/queue_sqlite/cycle/__init__.py b/src/queue_sqlite/cycle/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/queue_sqlite/cycle/listen_cycle.py b/src/queue_sqlite/cycle/listen_cycle.py new file mode 100644 index 0000000..e69de29 diff --git a/src/queue_sqlite/task_cycle/task_cycle.py b/src/queue_sqlite/cycle/task_cycle.py similarity index 100% rename from src/queue_sqlite/task_cycle/task_cycle.py rename to src/queue_sqlite/cycle/task_cycle.py diff --git a/src/queue_sqlite/task_cycle/__init__.py b/src/queue_sqlite/mounter/__init__.py similarity index 55% rename from src/queue_sqlite/task_cycle/__init__.py rename to src/queue_sqlite/mounter/__init__.py index db346d3..36b815d 100644 --- a/src/queue_sqlite/task_cycle/__init__.py +++ b/src/queue_sqlite/mounter/__init__.py @@ -1,5 +1,3 @@ from .task_mounter import TaskMounter -__all__ = [ - "TaskMounter" -] \ No newline at end of file +__all__ = ["TaskMounter"] \ No newline at end of file diff --git a/src/queue_sqlite/mounter/base_mounter.py b/src/queue_sqlite/mounter/base_mounter.py new file mode 100644 index 0000000..803ad53 --- /dev/null +++ b/src/queue_sqlite/mounter/base_mounter.py @@ -0,0 +1,42 @@ +from typing import Callable, List + + +class BaseMounter: + @classmethod + def mount_(cls, function: Callable): + setattr(cls, function.__name__, function) + + @staticmethod + def mount(meta: dict = {}): + """带参数的装饰器""" + def decorator(function: Callable): + # 使用自定义名称或函数原名 + function.meta = meta # type: ignore + setattr(BaseMounter, function.__name__, function) + return function + return decorator + + @classmethod + def get_function(cls, name: str): + return getattr(cls, name, None) + + @classmethod + def get_task_list(cls) -> List[str]: + """获取所有挂载的任务函数名称列表""" + task_list = [] + + # 遍历类属性字典 + for attr_name, attr_value in vars(cls).items(): + # 过滤条件： + # 1. 必须是可调用对象（函数） + # 2. 不是类自带的特殊方法（非双下划线开头） + # 3. 不是类方法本身（如 mount_, get_task_list 等） + if ( + callable(attr_value) + and not attr_name.startswith("__") + and attr_name not in ["mount_", "mount", "get_function", "get_task_list"] + ): + task_list.append(attr_name) + + return task_list + \ No newline at end of file diff --git a/src/queue_sqlite/mounter/listen_mounter.py b/src/queue_sqlite/mounter/listen_mounter.py new file mode 100644 index 0000000..2f411a8 --- /dev/null +++ b/src/queue_sqlite/mounter/listen_mounter.py @@ -0,0 +1,41 @@ +from typing import Callable, List + + +class ListenMounter: + @classmethod + def mount_Listener(cls, function: Callable): + setattr(cls, function.__name__, function) + + @staticmethod + def listener(): + """带参数的装饰器""" + def decorator(function: Callable): + # 使用自定义名称或函数原名 + setattr(ListenMounter, function.__name__, function) + return function + return decorator + + @classmethod + def get_Listener_function(cls, name: str): + return getattr(cls, name, None) + + @classmethod + def get_Listener_list(cls) -> List[str]: + """获取所有挂载的监听函数名称列表""" + listener_list = [] + + # 遍历类属性字典 + for attr_name, attr_value in vars(cls).items(): + # 过滤条件： + # 1. 必须是可调用对象（函数） + # 2. 不是类自带的特殊方法（非双下划线开头） + # 3. 不是类方法本身（如 mount_Listener, listener 等） + if ( + callable(attr_value) + and not attr_name.startswith("__") + and attr_name not in ["mount_Listener", "listener", "get_Listener_function", "get_Listener_list"] + ): + listener_list.append(attr_name) + + return listener_list + \ No newline at end of file diff --git a/src/queue_sqlite/task_cycle/task_mounter.py b/src/queue_sqlite/mounter/task_mounter.py similarity index 100% rename from src/queue_sqlite/task_cycle/task_mounter.py rename to src/queue_sqlite/mounter/task_mounter.py diff --git a/src/queue_sqlite/queue_operation/listen_operation.py b/src/queue_sqlite/queue_operation/listen_operation.py new file mode 100644 index 0000000..668d85f --- /dev/null +++ b/src/queue_sqlite/queue_operation/listen_operation.py @@ -0,0 +1,71 @@ +from ..mounter.listen_mounter import ListenMounter +import sqlite3 +import threading + + +class ListenOperation: + def __init__(self, db_dir): + self.db_dir = db_dir + self.listen_fields = ListenMounter.get_Listener_list() + self.create_table() + + def create_table(self): + if len(self.listen_fields) == 0: + return + sql = f""" + CREATE TABLE IF NOT EXISTS listen_table ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + key Text, + value Text + ); + """ + self.conn = sqlite3.connect( + self.db_dir, + check_same_thread = False, + ) + self.conn.execute(sql) + self.conn.execute("PRAGMA journal_mode=WAL;") + self.conn.execute("PRAGMA synchronous=NORMAL;") + self.conn.execute("PRAGMA cache_size=-20000;") + self.conn.execute("PRAGMA mmap_size=1073741824;") + self.conn.execute("PRAGMA temp_store=MEMORY;") + self.conn.execute("PRAGMA busy_timeout=5000;") + self.conn.commit() + + # 删除原有数据 + sql = f""" + DELETE FROM listen_table + """ + self.conn.execute(sql) + self.conn.commit() + + for listen_field in self.listen_fields: + sql = f""" + INSERT INTO + listen_table (key, value) + VALUES + (?, ?) + """ + self.conn.execute(sql, (listen_field, "null")) + self.conn.commit() + + def listen_data(self, key, value) -> tuple[bool, str]: + sql = f""" + Select value from listen_table where key = '{key}' + """ + cursor = self.conn.execute(sql) + result = cursor.fetchone() + if result is None: + return False, "key not found" + if result[0] == value: + return False, "value not changed" + else: + return True, result[0] + + def update_listen_data(self, key, value): + sql = f""" + UPDATE listen_table SET value = '{value}' WHERE key = '{key}' + """ + self.conn.execute(sql) + self.conn.commit() + diff --git a/src/queue_sqlite/queue_operation/queue_operation.py b/src/queue_sqlite/queue_operation/queue_operation.py index afa669b..1604384 100644 --- a/src/queue_sqlite/queue_operation/queue_operation.py +++ b/src/queue_sqlite/queue_operation/queue_operation.py @@ -8,7 +8,6 @@ from ..core import core from ..constant import MessageStatus - class QueueOperation(): def __init__(self, shard_num: int = 4, queue_name: str = "default"): self.shard_num = shard_num diff --git a/src/queue_sqlite/scheduler/__init__.py b/src/queue_sqlite/scheduler/__init__.py index 67cdd1f..71f369d 100644 --- a/src/queue_sqlite/scheduler/__init__.py +++ b/src/queue_sqlite/scheduler/__init__.py @@ -1,13 +1,20 @@ from .receive_scheduler import ReceiveScheduler from .task_scheduler import TaskScheduler +from .listen_data_scheduler import ListenDataScheduler from ..model import MessageItem from typing import Callable from ..queue_operation import QueueOperation +from ..queue_operation.listen_operation import ListenOperation from .cleanup_scheduler import CleanupScheduler +import os class QueueScheduler: def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): self.queue_operation = QueueOperation(shard_num) + self.listen_operation = ListenOperation(os.path.join(self.queue_operation.db_dir, "listen.db")) + self.listen_operation.create_table() + + self.listen_scheduler = ListenDataScheduler(self.listen_operation) self.receive_scheduler = ReceiveScheduler(self.queue_operation, receive_thread_num) self.task_scheduler = TaskScheduler(self.queue_operation, task_thread_num) self.cleanup_scheduler = CleanupScheduler(self.queue_operation) @@ -15,13 +22,18 @@ class QueueScheduler: def send_message(self, message: MessageItem, callback: Callable): self.receive_scheduler.send_message(message, callback) + def update_listen_data(self, key, value): + self.listen_operation.update_listen_data(key, value) + def start_queue_scheduler(self): self.receive_scheduler.start_receive_thread() self.task_scheduler.start_task_thread() self.cleanup_scheduler.start_cleanup() + self.listen_scheduler.start_listen_data() def stop_queue_scheduler(self): self.receive_scheduler.stop_receive_thread() + self.listen_scheduler.stop_listen_data() self.task_scheduler.stop_task_thread() self.cleanup_scheduler.stop_cleanup() # self.queue_operation.close_all_connections() diff --git a/src/queue_sqlite/scheduler/listen_data_scheduler.py b/src/queue_sqlite/scheduler/listen_data_scheduler.py new file mode 100644 index 0000000..b10ce8f --- /dev/null +++ b/src/queue_sqlite/scheduler/listen_data_scheduler.py @@ -0,0 +1,46 @@ +from ..queue_operation.listen_operation import ListenOperation +from concurrent.futures import ThreadPoolExecutor +from ..mounter.listen_mounter import ListenMounter +import threading + + +class ListenDataScheduler: + def __init__(self, listen_operation: ListenOperation): + self.listen_operation = listen_operation + self.is_running = False + self.executor = ThreadPoolExecutor(max_workers=1) + self.listen_thread = None + self.listen_data = {field: "null" for field in ListenMounter.get_Listener_list()} + + def _process_listen_data(self, key, value): + listen_function = ListenMounter.get_Listener_function(key) + if listen_function: + try: + listen_function(value) + except Exception as e: + ValueError(f"Error in {key} listener function: {e}") + + def listen(self): + while self.is_running: + for key, value in self.listen_data.items(): + status, new_value = self.listen_operation.listen_data(key, value) + if status: + self.listen_data[key] = new_value + self.executor.submit(self._process_listen_data, key, new_value) + + def start_listen_data(self): + if self.is_running: + return + self.is_running = True + self.listen_thread = threading.Thread(target=self.listen) + self.listen_thread.start() + + def stop_listen_data(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=2) + + self.executor.shutdown(wait=True) + \ No newline at end of file diff --git a/src/queue_sqlite/scheduler/task_scheduler.py b/src/queue_sqlite/scheduler/task_scheduler.py index 9927584..d638b34 100644 --- a/src/queue_sqlite/scheduler/task_scheduler.py +++ b/src/queue_sqlite/scheduler/task_scheduler.py @@ -1,8 +1,8 @@ from ..queue_operation import QueueOperation from ..model import MessageItem from ..constant import MessageStatus -from ..task_cycle.task_cycle import TaskCycle -from ..task_cycle.task_mounter import TaskMounter +from ..cycle.task_cycle import TaskCycle +from ..mounter.task_mounter import TaskMounter from concurrent.futures import ThreadPoolExecutor from datetime import datetime import time diff --git a/tests/listen/__init__.py b/tests/listen/__init__.py new file mode 100644 index 0000000..c0bba3e --- /dev/null +++ b/tests/listen/__init__.py @@ -0,0 +1,3 @@ +from .listen_data import key_1 + +__all__ = ["key_1"] \ No newline at end of file diff --git a/tests/listen/listen_data.py b/tests/listen/listen_data.py new file mode 100644 index 0000000..38dfcba --- /dev/null +++ b/tests/listen/listen_data.py @@ -0,0 +1,5 @@ +from queue_sqlite.mounter.listen_mounter import ListenMounter + +@ListenMounter.listener() +def key_1(data): + print(data) \ No newline at end of file diff --git a/tests/tasks/example.py b/tests/tasks/example.py index fc2954b..004dff2 100644 --- a/tests/tasks/example.py +++ b/tests/tasks/example.py @@ -1,4 +1,4 @@ -from queue_sqlite.task_cycle.task_mounter import TaskMounter +from queue_sqlite.mounter.task_mounter import TaskMounter from queue_sqlite.model import MessageItem diff --git a/tests/test_core.py b/tests/test_core.py index 2c3dc99..1eb16a9 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,5 +1,5 @@ from queue_sqlite.core import core -from queue_sqlite.task_cycle import TaskMounter +from queue_sqlite.mounter import TaskMounter from queue_sqlite.model import MessageItem diff --git a/tests/test_listen.py b/tests/test_listen.py new file mode 100644 index 0000000..f31f20a --- /dev/null +++ b/tests/test_listen.py @@ -0,0 +1,19 @@ +from listen import * +from queue_sqlite.scheduler import QueueScheduler +import time + + +class TestListen: + def test_listen_data(self): + scheduler = QueueScheduler() + scheduler.start_queue_scheduler() + scheduler.update_listen_data("key_1", "value_1") + time.sleep(0.001) + scheduler.update_listen_data("key_1", "value_2") + time.sleep(0.001) + scheduler.update_listen_data("key_1", "value_3") + time.sleep(0.001) + scheduler.update_listen_data("key_1", "value_4") + time.sleep(0.001) + scheduler.stop_queue_scheduler() + print(scheduler.listen_scheduler.listen_data)