代码之家 › 专栏 › 技术社区 › J. Taylor

当Asyncio.PriorityQueue为MaxSize且我放置()个新项时,如何将其从Asyncio.PriorityQueue中推出?

priority-queue python-asyncio queue python

J. Taylor · 技术社区 · 7 年前

我有一个 asyncio.PriorityQueue 我将其用作网络爬虫的URL队列,得分最低的URL在我调用时首先从队列中删除。 url_queue.get() . 当队列到达时 maxsize 项,默认行为是在调用时阻止 url_queue.put() ,直到呼叫 get() 从队列中删除一个项目以腾出空间。

我想做的是永远不要阻止,而是在我试图阻止的时候推掉得分最高的排队项目(或者至少是得分最高的项目之一)。 put() 得分较低的项目。有没有一种方法可以用这种方式自动从堆底部删除项 异步.PriorityQueue ?如果没有,是否有其他优先级队列/堆实现与Asyncio一起工作,这将使我能够做到这一点?或者其他一些数据结构/技术,可以让我拥有某种非阻塞的、优先级最高的队列?

谢谢!

1 回复 | 直到 7 年前

user4815162342 7 年前

有没有一种方法可以用这种方式自动从堆底部删除项 asyncio.PriorityQueue ?

默认情况下不是,但应该直接从继承 异步.PriorityQueue 只需实现所需的行为。与多线程队列实现不同,异步队列在单个线程中运行,因此不需要担心同步问题。

性能的一个可能问题是 PriorityQueue 不是设计为双端队列,因此它使用堆来存储项。堆可以是闽或 最大值 但不是两者都有;巨蟒的 heapq 模块实现最小堆,但您可以通过将优先级乘以-1来轻松模拟最大堆。在最小堆中,可以访问和弹出对数时间中最小的项,但不是最大的项,而在最大堆中,则相反。为了有效地处理最小和最大的项,您需要从继承 asyncio.Queue 并使用不同的数据结构来存储项,例如 sorted list .

例如(未测试):

class DroppingPriorityQueue(asyncio.Queue):
    def _init(self, maxsize):
        # called by asyncio.Queue.__init__
        self._queue = sortedcontainers.SortedList()

    def _put(self, item):
        # called by asyncio.Queue.put_nowait
        self._queue.add(item)

    def _get(self):
        # called by asyncio.Queue.get_nowait
        # pop the first (most important) item off the queue
        return self._queue.pop(0)

    def __drop(self):
        # drop the last (least important) item from the queue
        self._queue.pop()
        # no consumer will get a chance to process this item, so
        # we must decrement the unfinished count ourselves
        self.task_done()

    def put_nowait(self, item):
        if self.full():
            self.__drop()
        super().put_nowait(item)

    async def put(self, item):
        # Queue.put blocks when full, so we must override it.
        # Since our put_nowait never raises QueueFull, we can just
        # call it directly
        self.put_nowait(item)

类实现了两个不同的关注点:

它覆盖了 _get , _put 和 _init 使用的受保护方法 SortedList 作为底层存储。尽管未记录,但这些方法用于构建自定义队列,例如 PriorityQueue 和 LifoQueue 已经存在了几十年,首先是在 Queue 模块( queue 在python 3)中,稍后在 asyncio.queue .
它覆盖了 put 和 put_nowait 实现完整语义时丢弃的公共方法。