需要验证：可清除的 Python 队列类

Question

由于我不是Python和多线程编程方面的专家，我想问你我的实现是否正确。

我的目标是扩展 Queue 类，以便可以清除它。并应归还被移走的物品。就这样。我的实现是：

import Queue

class ClearableQueue(Queue.Queue):

    def __init__(self, maxsize):
        Queue.Queue.__init__(self, maxsize)

    def clear(self):
        self.mutex.acquire()

        copyOfRemovedEntries = list(self.queue)
        self.queue.clear()
        self.unfinished_tasks = 0
        self.all_tasks_done.notifyAll()
        self.not_full.notifyAll()

        self.mutex.release()

        return copyOfRemovedEntries

正确吗？ 谢谢。

更新：不幸的是，这个实现仍然不够，因为在调用clear()之后task_done可能会抛出ValueError异常。

更准确地说：队列被认为是在多线程环境中使用的。因此，假设一个生产者和一个工作线程（但您也可以考虑更多线程）。通常，如果工作线程调用 get()，则应在工作线程完成工作后调用 task_done()。如果这种情况发生，那么生产者线程可能会出于某种原因在工作线程调用 get() 之后、调用 task_done() 之前调用clear()。到目前为止，这是有效的，但是，如果工作线程想要调用task_done()，则会抛出异常。这是因为task_done()通过检查Queue类的unfinished_tasks来检查未完成任务的数量。

如果这个问题可以仅由 ClearableQueue 类处理，那么可以毫无顾虑地调用clear()方法，那将会很有趣。或者是否必须有不同的东西以某种方式控制方法调用。

实际上，在我的具体情况下，我不使用 join() 方法，因此我不需要调用 task_done()。不过，我想让这个功能更完整。对于其他人也可能有用。

Answer 1

如果您查看源代码，您将看到访问互斥锁的标准方法将变异代码包装在 try:finally 块中，以防出现问题：

import Queue

class ClearableQueue(Queue.Queue):

    def __init__(self, maxsize):
        Queue.Queue.__init__(self, maxsize)

    def clear(self):
        self.mutex.acquire()

        copyOfRemovedEntries = None
        try:
            copyOfRemovedEntries = list(self.queue)
            self.queue.clear()
            self.unfinished_tasks = 0
            self.all_tasks_done.notifyAll()
            self.not_full.notifyAll()
        finally:
            self.mutex.release()

        return copyOfRemovedEntries

编辑 1

如果您担心第二个线程抛出异常当您执行 get() 和 task_done() 时，为什么不将 task_done() 包装在 try-catch 块中呢？所有这些异常都告诉您，您已经确认了太多的项目，但是如果您的明确功能已经处理了它们，那么问题出在哪里？

如果它困扰您，这将隐藏该异常，使函数的意图更加明显，并删除我之前示例中的双重列表分配：

class ClearableQueue(Queue.Queue):

    def __init__(self, maxsize):
        Queue.Queue.__init__(self, maxsize)

    def get_all(self)
        self.mutex.acquire()

        try:
            copyOfRemovedEntries = list(self.queue)
            self.queue.clear()
            self.unfinished_tasks = 0
            self.all_tasks_done.notifyAll()
            self.not_full.notifyAll()
        finally:
            self.mutex.release()

        return copyOfRemovedEntries

    def clear(self):
        self.get_all()

    def task_done(self):
        try:
            Queue.Queue.task_done(self)
        except ValueError:
            pass

编辑 2

这作为一个更有效的解决方案怎么样？不隐藏任何内容：

class ClearableQueue(Queue.Queue):

    def __init__(self, maxsize):
        Queue.Queue.__init__(self, maxsize)
        self.tasks_cleared = 0

    def get_all(self)
        self.mutex.acquire()

        try:
            copyOfRemovedEntries = list(self.queue)
            self.queue.clear()
            self.unfinished_tasks = 0
            self.all_tasks_done.notifyAll()
            self.not_full.notifyAll()
            self.tasks_cleared += len(copyOfRemovedEntries)
        finally:
            self.mutex.release()

        return copyOfRemovedEntries

    def clear(self):
        self.get_all()

    def task_done(self):
        self.all_tasks_done.acquire()
        try:
            unfinished = self.unfinished_tasks + self.tasks_cleared - 1
            if unfinished <= 0:
                if unfinished < 0:
                    raise ValueError('task_done() called too many times')
                self.all_tasks_done.notify_all()
            self.unfinished_tasks = unfinished - self.tasks_cleared
            self.tasks_cleared = 0
        finally:
            self.all_tasks_done.release() 

我认为这应该避免异常，但仍然按照原始类的预期方式运行。

Answer 2

您似乎遇到了某种竞争条件，如果我理解的话，目前的情况是您有时会得到：

T1: |----->|------------->|-------------->|
    | get  |    some_opp  | task_done     |
T2: |---------->|------>|---------------->|
    | other_opp | clear | yet_another_opp |

Whereclear is Performed inside get and task_done。这会导致崩溃。据我了解，您需要某种方法来做到这一点：

T1: |----->|------------->|-------------->|
    | get  |    some_opp  | task_done     |
T2: |---------->|------------------------>|------>|
    | other_opp | wait_for_task_done      | clear |

如果这是正确的，您可能需要第二个锁，由 get 设置并由 task_done 释放，它表示“无法清除此队列”。然后，您可能需要一个 get 和 task_done 版本，在您真正知道自己在做什么的特殊情况下，它不会执行此操作。

另一种方法是拥有一个更原子的锁，它允许您执行此操作：

T1: |----->|------------------->|-------------->|------------->|
    | get  |    some_opp        | task_done     | finish_clear |
T2: |---------->|-------------->|---------------->|
    | other_opp | partial_clear | yet_another_opp |

您说“我还没有完成此任务，但您可以清除其余的任务”，然后告诉task_done该任务已尝试被清除，所以它应该在之后做一些事情。但这开始变得相当复杂。