高效管理Python队列：实现特定条件下最新元素替换的策略

pip install llist

import threading
from llist import dllist
from dataclasses import dataclass

# 1. 定义任务基类和特定任务类型
@dataclass
class Task:
    name: str

class UnimportantTask(Task):
    """表示非重要任务（B类任务）"""
    pass

# 2. 实现带有特定逻辑的队列管理器
class CustomQueue:
    def __init__(self):
        self.queue = dllist()  # 使用dllist作为底层队列
        self.unimportant_task_node = None  # 存储最新B类任务的节点引用
        self._lock = threading.Lock() # 用于保证线程安全

    def add(self, task):
        """
        向队列中添加任务。
        如果是UnimportantTask，会替换掉之前所有的UnimportantTask。
        """
        with self._lock: # 确保操作的原子性
            # 将新任务添加到队列尾部，并获取其节点引用
            new_node = self.queue.appendright(task)

            if isinstance(task, UnimportantTask):
                # 如果是B类任务，检查是否存在旧的B类任务
                if self.unimportant_task_node:
                    # 如果存在，则以O(1)复杂度移除旧的B类任务节点
                    self.queue.remove(self.unimportant_task_node)
                # 更新unimportant_task_node为新添加的B类任务的节点
                self.unimportant_task_node = new_node
            # 如果是A类任务，直接添加即可，unimportant_task_node保持不变

    def next(self):
        """
        从队列头部取出任务。
        """
        with self._lock: # 确保操作的原子性
            if not self.queue:
                return None # 队列为空

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

            # 如果取出的任务是当前记录的B类任务，则清除引用
            # 注意：这里需要比较节点对象，但dllist.popleft()返回的是数据，
            # 因此需要判断如果取出的任务是UnimportantTask，并且是唯一的那个，
            # 那么就清除unimportant_task_node引用。
            # 更严谨的做法是在add时存储(task, node)对，或在remove时判断
            # 但由于unimportant_task_node只存储最新的一个，
            # 只要pop出的task是UnimportantTask，且unimportant_task_node指向它，
            # 那么就清空。
            if isinstance(popped_task, UnimportantTask) and \
               self.unimportant_task_node and \
               self.unimportant_task_node.value is popped_task:
                self.unimportant_task_node = None

            return popped_task

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

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

# 创建自定义队列实例
tasks = CustomQueue()

# 添加A类任务
tasks.add(Task('A1'))
tasks.add(Task('A2'))

# 添加第一个B类任务
tasks.add(UnimportantTask('B1')) # B1进入队列

# 添加A类任务
tasks.add(Task('A3'))

# 添加第二个B类任务，此时B1应该被移除
tasks.add(UnimportantTask('B2')) # B1被移除，B2进入队列

# 添加第三个B类任务，此时B2应该被移除
tasks.add(UnimportantTask('B3')) # B2被移除，B3进入队列

# 添加A类任务
tasks.add(Task('A4'))

print("队列中的任务消费顺序：")
while task := tasks.next():
    print(task)

队列中的任务消费顺序：
Task(name='A1')
Task(name='A2')
Task(name='A3')
UnimportantTask(name='B3')
Task(name='A4')