modify: 协程调度器

src/core/src/queue_operation.rs

@@ -461,6 +461,31 @@ impl QueueOperation {
         Ok(())
     }
 
+    pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> {
+        let now: DateTime<Utc> = Utc::now();
+        let expire_time = now - Duration::days(days);
+        let iso_format = expire_time.to_rfc3339();
+
+        let conn = self
+            .pool
+            .get()
+            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+
+        conn.execute(
+            "
+            UPDATE messages 
+            SET is_deleted = 1 
+            WHERE is_deleted = 0
+            AND expire_time IS NOT NULL
+            AND expire_time < ?1
+            ",
+            params![iso_format],
+        )
+        .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+
+        Ok(())
+    }
+
     pub fn clean_expired_messages(&self) -> PyResult<()> {
         let now: DateTime<Utc> = Utc::now();
         let iso_format = now.to_rfc3339();

src/queue_sqlite/constant/__init__.py

@@ -2,4 +2,4 @@ from .message_priority import MessagePriority
 from .message_status import MessageStatus
 from .message_type import MessageType
 
-__all__ = ["MessagePriority", "MessageStatus", "MessageType"]
+__all__ = ["MessagePriority", "MessageStatus", "MessageType"]

src/queue_sqlite/constant/message_status.py

@@ -7,4 +7,3 @@ class MessageStatus(IntEnum):
     COMPLETED = 2
     FAILED = 3
     RETRYING = 4
-

src/queue_sqlite/constant/message_type.py

@@ -2,5 +2,5 @@ from enum import Enum
 
 
 class MessageType(Enum):
-    TASK = "task" # 任务
-    LISTEN = "listen" # 监听
+    TASK = "task"  # 任务
+    LISTEN = "listen"  # 监听

src/queue_sqlite/core/__init__.pyi

@@ -44,6 +44,7 @@ class core:
         def delete_message(self, message_id: str): ...
         def clean_old_messages(self, days: int): ...
         def clean_expired_messages(self): ...
+        def remove_expired_messages(self, days: int): ...
 
     class TaskMounter:
         def __init__(self, task_class: type[TaskMounter]): ...

src/queue_sqlite/cycle/async_task_cycle.py

@@ -0,0 +1,97 @@
+from ..model import MessageItem
+from ..constant import MessageStatus
+import json
+import asyncio
+import functools
+
+
+def retry_async(max_retries=3, delay=1):
+    """
+    异步重试装饰器
+
+    Args:
+        max_retries: 最大重试次数
+        delay: 重试间隔（秒）
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(self, *args, **kwargs):
+            last_exception = None
+            # 使用message_item中的retry_count作为最大重试次数
+            retries = getattr(self.message_item, "retry_count", max_retries)
+
+            # 至少尝试一次（retries+1），最多尝试max_retries+1次
+            actual_retries = min(retries, max_retries) if max_retries > 0 else retries
+
+            for attempt in range(actual_retries + 1):
+                try:
+                    return await func(self, *args, **kwargs)
+                except Exception as e:
+                    last_exception = e
+                    if attempt < actual_retries:
+                        await asyncio.sleep(delay)
+                    else:
+                        break
+            if last_exception:
+                raise last_exception
+            else:
+                raise Exception("Unknown error occurred during async task execution")
+
+        return wrapper
+
+    return decorator
+
+
+class AsyncTaskCycle:
+    def __init__(self, message_item: MessageItem, callback):
+        self.message_item = message_item
+        self.callback = callback
+        self.task_result = None
+        self.task_status = None
+        self.task_error = None
+
+    @retry_async(max_retries=3, delay=1)
+    async def run(self):
+        try:
+            task_result = await self.callback(self.message_item)  # type: ignore
+        except Exception as e:
+            self.task_result = None
+            self.task_status = MessageStatus.FAILED
+            self.task_error = str(e)
+        else:
+            self.task_result = task_result
+            self.task_status = MessageStatus.COMPLETED
+            self.task_error = None
+
+    def get_task_result(self):
+        if isinstance(self.task_result, (dict, list)):
+            try:
+                return json.dumps(self.task_result)
+            except:
+                return json.dumps({"result": str(self.task_result)})
+
+        elif isinstance(self.task_result, str):
+            try:
+                json.loads(self.task_result)
+                return self.task_result
+            except:
+                return json.dumps({"result": self.task_result})
+        elif isinstance(self.task_result, (int, float, bool)):
+            return json.dumps({"result": self.task_result})
+        elif self.task_result is None:
+            return "null"
+        else:
+            return json.dumps({"result": str(self.task_result)})
+
+    def get_task_status(self):
+        return self.task_status
+
+    def get_task_error(self):
+        return self.task_error
+
+    def get_task_message_item(self):
+        return self.message_item
+
+    def get_task_callback(self):
+        return self.callback

src/queue_sqlite/cycle/task_cycle.py

@@ -3,6 +3,50 @@ from typing import Callable, Optional
 from ..constant import MessageStatus
 import json
 
+import functools
+import time
+
+
+def retry_sync(max_retries=3, delay=1):
+    """
+    同步重试装饰器
+
+    Args:
+        max_retries: 最大重试次数
+        delay: 重试间隔（秒）
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(self, *args, **kwargs):
+            last_exception = None
+            # 使用message_item中的retry_count作为最大重试次数
+            retries = getattr(self.message_item, "retry_count", max_retries)
+
+            # 至少尝试一次（retries+1），最多尝试max_retries+1次
+            actual_retries = min(retries, max_retries) if max_retries > 0 else retries
+
+            for attempt in range(actual_retries + 1):
+                try:
+                    return func(self, *args, **kwargs)
+                except Exception as e:
+                    last_exception = e
+                    if attempt < actual_retries:
+                        time.sleep(delay)
+                    else:
+                        break
+
+            # 如果没有捕获到异常，则创建一个新的异常
+            if last_exception is not None:
+                raise last_exception
+            else:
+                raise Exception("Unknown error occurred during sync task execution")
+
+        return wrapper
+
+    return decorator
+
+
 class TaskCycle:
     def __init__(self, message_item: MessageItem, callback: Optional[Callable]):
         self.message_item = message_item
@@ -11,15 +55,16 @@ class TaskCycle:
         self.task_status = None
         self.task_error = None
 
+    @retry_sync(max_retries=3, delay=1)
     def run(self):
         try:
-            task_result = self.callback(self.message_item) # type: ignore
+            task_result = self.callback(self.message_item)  # type: ignore
         except Exception as e:
             self.task_result = None
             self.task_status = MessageStatus.FAILED
             self.task_error = str(e)
         else:
-            self.task_result = task_result # type: ignore
+            self.task_result = task_result  # type: ignore
             self.task_status = MessageStatus.COMPLETED
             self.task_error = None
 
@@ -28,31 +73,29 @@ class TaskCycle:
             try:
                 return json.dumps(self.task_result)
             except:
-                return json.dumps({'result': str(self.task_result)})     
+                return json.dumps({"result": str(self.task_result)})
 
         elif isinstance(self.task_result, str):
             try:
                 json.loads(self.task_result)
                 return self.task_result
             except:
-                return json.dumps({'result': self.task_result})
+                return json.dumps({"result": self.task_result})
         elif isinstance(self.task_result, (int, float, bool)):
-            return json.dumps({'result': self.task_result})
+            return json.dumps({"result": self.task_result})
         elif self.task_result is None:
-            return 'null'
+            return "null"
         else:
-            return json.dumps({'result': str(self.task_result)})
+            return json.dumps({"result": str(self.task_result)})
 
     def get_task_status(self):
         return self.task_status
 
     def get_task_error(self):
         return self.task_error
-    
+
     def get_task_message_item(self):
         return self.message_item
-    
+
     def get_task_callback(self):
         return self.callback
-    
-    

src/queue_sqlite/queue_operation/listen_operation.py

@@ -19,12 +19,21 @@ class ListenOperation:
     def create_table(self):
         if len(self.listen_fields) == 0:
             return
-        sql = f"""
+        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,
@@ -34,18 +43,28 @@ class ListenOperation:
                 new_value TEXT,
                 is_delete integer DEFAULT 0,
                 timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
-            );
-            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;
+            )
         """
-        conn = self._get_connection()
-        conn.executescript(sql)
+        )
+
+        # 检查触发器是否已存在，如果不存在则创建
+        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.execute("PRAGMA journal_mode=WAL;")
         conn.execute("PRAGMA synchronous=NORMAL;")
         conn.execute("PRAGMA cache_size=-20000;")
@@ -54,15 +73,8 @@ class ListenOperation:
         conn.execute("PRAGMA busy_timeout=5000;")
         conn.commit()
 
-        # 删除原有数据
-        # sql = f"""
-        #     DELETE FROM listen_table
-        # """
-        # conn.execute(sql)
-        # conn.commit()
-
         for listen_field in self.listen_fields:
-            sql = f"""
+            sql = """
                 INSERT INTO 
                     listen_table (key, value)
                 VALUES 

src/queue_sqlite/queue_operation/queue_operation.py

@@ -120,3 +120,8 @@ class QueueOperation:
     def clean_expired_messages(self, shard_index: int):
         conn = self._get_shard_conn(shard_index)
         conn.clean_expired_messages()
+
+    # 彻底删除消息
+    def remove_expired_messages(self, shard_index: int, remove_days: int = 30):
+        conn = self._get_shard_conn(shard_index)
+        conn.remove_expired_messages(remove_days)

src/queue_sqlite/scheduler/__init__.py

@@ -1,24 +1,27 @@
 from .standard import StandardQueueScheduler
+from ._async import AsyncQueueScheduler
 from .base import BaseScheduler
 from ..model import MessageItem
 from typing import Callable
+import multiprocessing
+
+
+SCHEDULER_TYPES = {"standard": StandardQueueScheduler, "async": AsyncQueueScheduler}
 
 
 class QueueScheduler(BaseScheduler):
     def __init__(
         self,
         receive_thread_num=1,
-        task_thread_num=1,
+        task_thread_num=multiprocessing.cpu_count() * 2,
         shard_num=4,
         scheduler_type="standard",
     ):
-        self.scheduler = None
-        if scheduler_type == "standard":
-            self.scheduler = StandardQueueScheduler(
-                receive_thread_num, task_thread_num, shard_num
-            )
-        else:
-            pass
+        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)
+
         if self.scheduler:
             self.queue_operation = self.scheduler.queue_operation
 

src/queue_sqlite/scheduler/_async/__init__.py

@@ -0,0 +1,49 @@
+from .async_receive_scheduler import AsyncReceiveScheduler
+from .async_task_scheduler import AsyncTaskScheduler
+from .async_listen_data_scheduler import AsyncListenDataScheduler
+from ...model import MessageItem
+from ...queue_operation.listen_operation import ListenOperation
+from typing import Callable
+from ...queue_operation import QueueOperation
+import os
+from ..base import BaseScheduler
+from ..cleanup_scheduler import CleanupScheduler
+
+
+class AsyncQueueScheduler(BaseScheduler):
+
+    def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4):
+        self.queue_operation = QueueOperation(shard_num=shard_num)
+        self.receive_scheduler = AsyncReceiveScheduler(
+            self.queue_operation, 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.cleanup_scheduler = CleanupScheduler(self.queue_operation)
+
+    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 get_listen_datas(self):
+        return self.listen_operation.get_values()
+
+    def get_listen_data(self, key):
+        return self.listen_operation.get_value(key)
+
+    def start(self):
+        self.receive_scheduler.start_receive_thread()
+        self.task_scheduler.start_task_thread()
+
+    def stop(self):
+        self.receive_scheduler.stop_receive_thread()
+        self.task_scheduler.stop_task_thread()
+
+
+__all__ = ["AsyncQueueScheduler"]

src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py

@@ -0,0 +1,67 @@
+from ...queue_operation.listen_operation import ListenOperation
+import asyncio
+from ...mounter.listen_mounter import ListenMounter
+import threading
+import multiprocessing
+import concurrent.futures
+
+
+class AsyncListenDataScheduler:
+    def __init__(self, listen_operation: ListenOperation):
+        self.listen_operation = listen_operation
+        self.is_running = False
+        self.thread_num = multiprocessing.cpu_count()
+        self.listen_thread = None
+
+    async def _process_listen_data(self, key, value, delete_id):
+        listen_function = ListenMounter.get_Listener_function(key)
+        if listen_function:
+            if asyncio.iscoroutinefunction(listen_function):
+                await listen_function(value)
+            else:
+                try:
+                    loop = asyncio.get_event_loop()
+                    with concurrent.futures.ThreadPoolExecutor(
+                        max_workers=self.thread_num
+                    ) as executor:
+                        await loop.run_in_executor(executor, listen_function, value)
+                    listen_function(value)
+                except Exception as e:
+                    ValueError(f"Error in {key} listener function: {e}")
+                finally:
+                    self.listen_operation.delete_change_log(delete_id=delete_id)
+
+    async def listen(self):
+        async with asyncio.Semaphore(self.thread_num):
+            while self.is_running:
+                status, change_data_items = self.listen_operation.listen_data()
+                tasks = []
+                if status:
+                    for data in change_data_items:
+                        key = data[6]
+                        new_value = data[7]
+                        delete_id = data[0]
+                        tasks.append(
+                            asyncio.create_task(
+                                self._process_listen_data(key, new_value, delete_id)
+                            )
+                        )
+                    if tasks:
+                        await asyncio.gather(*tasks, return_exceptions=True)
+
+                else:
+                    await asyncio.sleep(0.05)
+
+    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)

src/queue_sqlite/scheduler/_async/async_receive_scheduler.py

@@ -0,0 +1,102 @@
+from ...queue_operation import QueueOperation
+from ...model import MessageItem
+from typing import Callable
+import asyncio
+import threading
+import concurrent.futures
+
+
+class AsyncReceiveScheduler:
+    def __init__(self, queue_operation: QueueOperation, receive_thread_num: int = 1):
+        self.callbacks = dict()
+        self.receive_thread_num = receive_thread_num
+        self.is_running = False
+        self.lock = threading.Lock()
+        self.queue_operation = queue_operation
+        self.receive_thread = None  # 单一轮询线程
+
+    def send_message(self, message: MessageItem, callback: Callable = None):  # type: ignore
+        if callback is None:
+            callback = lambda message: print(message)
+        # self.queue_operation.enqueue(message.to_dict())
+        self.queue_operation.enqueue(message.to_dict_by_core())
+        with self.lock:
+            self.callbacks[message.id] = callback
+
+    async def receive_message(self):
+        """单一轮询线程，并行执行回调"""
+
+        def callback_default(message):
+            return message
+
+        async with asyncio.Semaphore(self.receive_thread_num):
+            while self.is_running:
+                message_list = self.queue_operation.get_completed_messages()
+                tasks = []
+                if message_list:
+                    for message in message_list:
+                        message = MessageItem.from_dict(message)
+                        with self.lock:
+                            callback = self.callbacks.pop(message.id, None)
+                            if callback is None:
+                                callback = callback_default
+
+                        tasks.append(
+                            asyncio.create_task(self._safe_callback(callback, message))
+                        )
+                    if tasks:
+                        await asyncio.gather(*tasks, return_exceptions=True)
+                else:
+                    await asyncio.sleep(0.05)
+
+    async def _safe_callback(self, callback, message):
+        """安全执行回调函数"""
+        try:
+            # 检查回调函数是否是异步函数
+            if asyncio.iscoroutinefunction(callback):
+                await callback(message)
+            else:
+                # 如果是同步函数，则在执行器中运行以避免阻塞事件循环
+                loop = asyncio.get_event_loop()
+                with concurrent.futures.ThreadPoolExecutor(
+                    max_workers=self.receive_thread_num
+                ) as executor:
+                    await loop.run_in_executor(executor, callback, message)
+        except Exception as e:
+            print(f"回调执行错误: {str(e)}")
+        finally:
+            self.queue_operation.delete_message(message.id)
+
+    def _run_receive_loop(self):
+        """在新事件循环中运行接收消息循环"""
+        # 创建新的事件循环
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        try:
+            # 运行异步接收函数直到停止
+            loop.run_until_complete(self.receive_message())
+        except Exception as e:
+            print(f"任务调度出错: {str(e)}")
+        finally:
+            loop.close()
+
+    def start_receive_thread(self):
+        if self.is_running:
+            return
+
+        self.is_running = True
+        # 创建单一轮询线程
+        self.receive_thread = threading.Thread(
+            target=self._run_receive_loop, daemon=True
+        )
+        self.receive_thread.start()
+
+    def stop_receive_thread(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=5.0)

src/queue_sqlite/scheduler/_async/async_task_scheduler.py

@@ -0,0 +1,137 @@
+from ...queue_operation import QueueOperation
+from ...model import MessageItem
+from ...constant import MessageStatus
+from ...cycle.async_task_cycle import AsyncTaskCycle
+from ...mounter.task_mounter import TaskMounter
+import asyncio
+from datetime import datetime
+import threading
+import multiprocessing
+
+
+class AsyncTaskScheduler:
+
+    def __init__(
+        self,
+        queue_operation: QueueOperation,
+        task_thread_num: int = multiprocessing.cpu_count() * 2,
+    ):
+        self.task_thread_num = task_thread_num
+        self.is_running = False
+        self.queue_operation = queue_operation
+        self.task_thread = None  # 单一轮询线程
+
+    async def _process_message(self, message: MessageItem):
+        """处理单个消息"""
+        try:
+            if message.destination == "client":
+                message.status = MessageStatus.COMPLETED
+                message.result = {"result": "success"}
+                message.updatetime = datetime.now()
+                return message
+
+            task_function = TaskMounter.get_task_function(message.destination)
+            if task_function is None:
+                raise ValueError(
+                    f"Task function not found for destination: {message.destination}"
+                )
+
+            task_cycle = AsyncTaskCycle(message, task_function)
+            await task_cycle.run()
+
+            message.status = task_cycle.get_task_status()  # type: ignore
+            if message.status == MessageStatus.FAILED:
+                message.result = {"error": task_cycle.get_task_error()}
+            else:
+                message.result = task_cycle.get_task_result()  # type: ignore
+            message.updatetime = datetime.now()
+        except Exception as e:
+            message.status = MessageStatus.FAILED
+            message.result = {"error": str(e)}
+            message.updatetime = datetime.now()
+
+        return message
+
+    def _process_messages(self, message):
+        """更新任务结果到数据库"""
+        try:
+            self.queue_operation.update_result(message.id, message.result)
+            self.queue_operation.update_status(message.id, message.status)
+        except Exception as e:
+            print(f"任务结果更新失败: {str(e)}")
+
+    async def task_callback(self):
+        """单一轮询线程，并行执行任务"""
+        # 使用信号量控制并发任务数
+        semaphore = asyncio.Semaphore(self.task_thread_num)
+
+        async def limited_process_message(message):
+            async with semaphore:
+                return await self._process_message(MessageItem.from_dict(message))
+
+        while self.is_running:
+            try:
+                message_list = self.queue_operation.dequeue(
+                    size=self.task_thread_num * 2
+                )
+                if not message_list:
+                    # 使用指数退避策略减少轮询开销
+                    await asyncio.sleep(0.05)  # 增加延迟减少CPU使用
+                    continue
+
+                # 并发处理消息
+                tasks = [
+                    asyncio.create_task(limited_process_message(message))
+                    for message in message_list
+                ]
+
+                # 等待所有任务完成
+                completed, _ = await asyncio.wait(
+                    tasks,
+                    return_when=asyncio.ALL_COMPLETED,
+                    timeout=30.0,  # 添加超时防止无限等待
+                )
+
+                # 批量处理结果
+                for task in completed:
+                    try:
+                        message = task.result()
+                        self._process_messages(message)
+                    except Exception as e:
+                        print(f"处理任务结果时出错: {str(e)}")
+
+            except Exception as e:
+                print(f"任务调度出错: {str(e)}")
+                await asyncio.sleep(0.1)  # 出错时短暂延迟
+
+    def _run_task_loop(self):
+        """在新事件循环中运行任务处理循环"""
+        # 创建新的事件循环
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        try:
+            # 运行异步任务处理函数直到停止
+            loop.run_until_complete(self.task_callback())
+        except Exception as e:
+            print(f"任务调度出错: {str(e)}")
+        finally:
+            loop.close()
+
+    def start_task_thread(self):
+        if self.is_running:
+            return
+
+        self.is_running = True
+        self.task_thread = threading.Thread(target=self._run_task_loop)
+        self.task_thread.daemon = True
+        self.task_thread.start()
+
+    def stop_task_thread(self):
+        if not self.is_running:
+            return
+
+        self.is_running = False
+        if self.task_thread:
+            self.task_thread.join(timeout=5.0)
+            self.task_thread = None

src/queue_sqlite/scheduler/cleanup_scheduler.py

@@ -1,26 +1,31 @@
 # 新建清理调度器
 import threading
 import time
-from ...queue_operation import QueueOperation
+from ..queue_operation import QueueOperation
 
 
 class CleanupScheduler:
-    def __init__(self, queue_operation: QueueOperation, interval_minutes=60):
+    def __init__(
+        self, queue_operation: QueueOperation, interval_minutes=60, remove_days=30
+    ):
         self.queue_operation = queue_operation
         self.interval = interval_minutes * 60  # 转换为秒
         self.is_running = False
         self.cleanup_thread = None
+        self.remove_days = remove_days
+
+        for i in range(self.queue_operation.shard_num):
+            # 清理过期但未处理的消息
+            self.queue_operation.clean_expired_messages(i)
+            # 彻底删除30天前的消息
+            self.queue_operation.remove_expired_messages(i, self.remove_days)
 
     def cleanup_expired_messages(self):
         """清理过期消息"""
         while self.is_running:
             try:
-                # 清理7天前的已完成/失败消息
-                for i in range(self.queue_operation.shard_num):
-                    self.queue_operation.clean_old_messages(i, 7)
-
-                # 清理过期但未处理的消息
                 for i in range(self.queue_operation.shard_num):
+                    # 清理过期但未处理的消息
                     self.queue_operation.clean_expired_messages(i)
 
             except Exception as e:

src/queue_sqlite/scheduler/standard/__init__.py

@@ -5,7 +5,7 @@ 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
+from ..cleanup_scheduler import CleanupScheduler
 import os
 from ..base import BaseScheduler
 

src/queue_sqlite/scheduler/standard/listen_data_scheduler.py

@@ -2,13 +2,14 @@ from ...queue_operation.listen_operation import ListenOperation
 from concurrent.futures import ThreadPoolExecutor
 from ...mounter.listen_mounter import ListenMounter
 import threading
+import multiprocessing
 
 
 class ListenDataScheduler:
     def __init__(self, listen_operation: ListenOperation):
         self.listen_operation = listen_operation
         self.is_running = False
-        self.executor = ThreadPoolExecutor(max_workers=1)
+        self.executor = ThreadPoolExecutor(max_workers=multiprocessing.cpu_count())
         self.listen_thread = None
 
     def _process_listen_data(self, key, value, delete_id):

src/queue_sqlite/scheduler/standard/task_scheduler.py

@@ -7,10 +7,15 @@ from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 import time
 import threading
+import multiprocessing
 
 
 class TaskScheduler:
-    def __init__(self, queue_operation: QueueOperation, task_thread_num: int = 2):
+    def __init__(
+        self,
+        queue_operation: QueueOperation,
+        task_thread_num: int = multiprocessing.cpu_count() * 2,
+    ):
         self.task_thread_num = task_thread_num
         self.is_running = False
         self.executor = ThreadPoolExecutor(max_workers=task_thread_num)  # 并行执行任务