Python中实现带条件淘汰机制的队列：基于双向链表的最新元素管理策略

from llist import dllist
from dataclasses import dataclass
import threading # 用于线程安全考虑

@dataclass
class Task:
    """定义基础任务类"""
    name: str

class UnimportantTask(Task):
    """定义非重要任务类，继承自Task"""
    pass

class CustomQueue:
    """
    实现带有条件淘汰机制的定制化队列。
    非重要任务(UnimportantTask)只保留最新一个，重要任务(Task)全部保留。
    """
    def __init__(self):
        self.queue = dllist()  # 使用dllist作为底层队列
        self.unimportant_task_node = None # 存储最新非重要任务的节点引用
        self._lock = threading.Lock() # 用于保证线程安全

    def add(self, task: Task):
        """
        向队列中添加任务。
        如果是非重要任务，会先移除队列中现有的非重要任务。
        """
        with self._lock: # 确保操作的原子性
            # 将新任务添加到队列末尾，并获取其节点引用
            new_node = self.queue.appendright(task)

            if isinstance(task, UnimportantTask):
                # 如果是新的非重要任务
                if self.unimportant_task_node:
                    # 如果队列中已经存在旧的非重要任务，则移除它
                    self.queue.remove(self.unimportant_task_node)
                # 更新引用，指向最新的非重要任务节点
                self.unimportant_task_node = new_node

    def next(self) -> Task | None:
        """
        从队列头部取出下一个任务。
        如果队列为空，返回None。
        """
        with self._lock: # 确保操作的原子性
            if not self.queue:
                return None

            # 从队列头部取出任务
            task = self.queue.popleft()

            if isinstance(task, UnimportantTask):
                # 如果取出的任务是非重要任务，说明它已经离开了队列
                # 因此需要清空对应的节点引用
                self.unimportant_task_node = None
            return task

    def __len__(self):
        """返回队列当前长度"""
        with self._lock:
            return len(self.queue)

    def __bool__(self):
        """判断队列是否为空"""
        with self._lock:
            return bool(self.queue)

# 演示代码
if __name__ == "__main__":
    tasks_queue = CustomQueue()

    print("--- 添加任务 ---")
    tasks_queue.add(Task('A1'))
    tasks_queue.add(Task('A2'))
    tasks_queue.add(UnimportantTask('B1')) # B1进入，如果之前有B会被移除
    tasks_queue.add(Task('A3'))
    tasks_queue.add(UnimportantTask('B2')) # B2进入，B1被移除
    tasks_queue.add(UnimportantTask('B3')) # B3进入，B2被移除
    tasks_queue.add(Task('A4'))

    print(f"队列当前长度: {len(tasks_queue)}")

    print("\n--- 取出任务 ---")
    while task := tasks_queue.next():
        print(task)

    print(f"\n队列最终长度: {len(tasks_queue)}")

--- 添加任务 ---
队列当前长度: 5

--- 取出任务 ---
Task(name='A1')
Task(name='A2')
Task(name='A3')
UnimportantTask(name='B3')
Task(name='A4')

队列最终长度: 0