first commit

tests/tasks/__init__.py

@@ -0,0 +1,5 @@
+from .example import example
+
+__all__ = [
+    "example"
+]

tests/tasks/example.py

@@ -0,0 +1,19 @@
+from queue_sqlite.task_cycle.task_mounter import TaskMounter
+from queue_sqlite.model import MessageItem
+
+
+@TaskMounter.task(meta={"task_name": "test"})
+def example(message_item: MessageItem):
+    def fibonacci_generator():
+        a, b = 0, 1
+        while True:
+            yield a
+            a, b = b, a + b
+
+    # 示例：获取前10项
+    fib = fibonacci_generator()
+    message_item.result = {"fibonacci": [next(fib) for _ in range(500)]}
+    return message_item.to_json()
+
+
+# 输出：[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

tests/test_core.py

@@ -0,0 +1,31 @@
+from queue_sqlite.core import core
+from queue_sqlite.task_cycle import TaskMounter
+from queue_sqlite.model import MessageItem
+
+
+class TestCore:
+    queue_operation = core.QueueOperation("./cache/test.db")
+
+    @classmethod
+    def test_task_mounter(cls):
+        task_mounter = core.TaskMounter(TaskMounter)
+        task_mounter.get_task_list()
+        print(task_mounter.get_task_function("task")({"num": 1})(lambda x: x+1).meta["num"])
+        from tasks import example
+        task_mounter.get_task_list()
+        print(task_mounter.get_task_function("<lambda>"))
+    
+    @classmethod
+    def test_queue_operation_init_db(cls):
+        cls.queue_operation.init_db()
+
+    @classmethod
+    def test_queue_operation_enqueue(cls):
+        message_item = MessageItem(content={"num": 1}, destination="test")
+        cls.queue_operation.enqueue(message_item.to_dict_by_core())
+
+    @classmethod
+    def test_queue_operation_dequeue(cls):
+        message_item = MessageItem(content={"num": 1}, destination="test")
+        cls.queue_operation.enqueue(message_item.to_dict_by_core())
+        print(cls.queue_operation.dequeue(1))

tests/test_message_item.py

@@ -0,0 +1,18 @@
+from queue_sqlite.model import MessageItem
+
+
+class TestMessageItem:
+    messageItem = MessageItem(content={"num": 1}, destination="test")
+    
+    @classmethod
+    def test_to_dict(cls):
+        print(cls.messageItem.to_dict())
+
+    @classmethod
+    def test_to_json(cls):
+        print(cls.messageItem.to_json())
+
+    @classmethod
+    def test_from_dict(cls):
+        messageItem = MessageItem.from_dict(cls.messageItem.to_dict())
+        print(messageItem.to_dict())

tests/test_stress.py

@@ -0,0 +1,80 @@
+from queue_sqlite.model import MessageItem
+from queue_sqlite.scheduler import QueueScheduler
+import time
+from tasks import *
+import threading
+import psutil
+
+
+class TestStress:
+
+    callback_counter = 0
+    lock = threading.Lock()
+    is_monitoring = False
+
+    @classmethod
+    def monitor_resources(cls):
+        """监控资源使用情况"""
+        while cls.is_monitoring:
+            cpu = psutil.cpu_percent()
+            memory = psutil.virtual_memory().percent
+            print(f"CPU使用率: {cpu}%, 内存使用率: {memory}%")
+            time.sleep(0.5)
+
+    @classmethod
+    def _callback(cls, message_item: MessageItem):
+        with cls.lock:
+            cls.callback_counter += 1
+        # print(f"callback: {message_item}")
+        # print(f"callback: {message_item.id}")
+        # print(f"callback: {message_item.expire_time}")
+        # print(f"callback: {message_item.tags}")
+
+    @classmethod
+    def test_stress(cls):
+        TASK_NUM = 1000
+        messages = []
+        for i in range(TASK_NUM):  # 增加任务数量
+            message_item = MessageItem(content={"num": i}, destination="example")
+            messages.append(message_item)
+        
+        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)
+        
+        # threads = []
+        # for message_item in messages:
+        #     t = threading.Thread(target=queue_scheduler.send_message, args=(message_item, cls.test_callback))
+        #     threads.append(t)
+
+        # 添加性能计时
+        start_time = time.perf_counter()
+        
+        queue_scheduler.start_queue_scheduler()
+        
+        for message_item in messages:  # 增加任务数量
+            message_item: MessageItem
+            queue_scheduler.send_message(message_item, cls._callback)
+        
+        # # 多线程发送任务
+        # for t in threads:
+        #     t.start()
+        
+        # # 动态等待任务完成
+        start_wait = time.time()
+        while (queue_scheduler.queue_operation.get_queue_length() > 0 or cls.callback_counter < TASK_NUM):  # 30秒超时
+            time.sleep(0.5)
+            print(f"当前队列长度：{queue_scheduler.queue_operation.get_queue_length()}")
+            print(f"当前回调数量：{cls.callback_counter}")
+        
+        queue_scheduler.stop_queue_scheduler()
+        
+        print(f"当前队列长度：{queue_scheduler.queue_operation.get_queue_length()}")
+        print(f"测试函数：斐波那契数列")
+        print(f"次数：{TASK_NUM}")
+        print(f"总耗时: {time.perf_counter() - start_time:.2f}秒")
+
+        cls.is_monitoring = False
+        monitor_thread.join(timeout=1)
+