如何使 Queue 子类的 putall 和 getall 方法成为原子方法？

Question

我有一个 class 继承自 Queue，并被另一个工厂使用 class:

class myQueue(Queue):
    def putall(self , mobjects , *args):
        self.put(mobject, *args )

    def getall(self , number , *args ):
        return [self.get(*args) for _ in xrange(number)]

我在外部使用 putall 和 getall，并希望它们是原子的。我怎样才能做到这一点？我正在使用 Python 2.6.

Answer 1

为了使 putall 和 getall 成为原子，您需要持有一个锁，以防止任何其他方法从运行修改 Queue。显而易见的选择是使用 Queue 内部使用的 mutex 对象，但它是使用 threading.Lock 实现的，这意味着它不能递归使用。这意味着天真的解决方案将导致死锁：

from Queue import Queue
import threading

class myQueue(Queue):
    def putall(self , objects , *args):
        with self.mutex:
            for object in objects:
                self.put(object, *args) # This will hang.

    def getall(self , number , *args ):
        with self.mutex:
            return [self.get(*args) for _ in xrange(number)] # This will hang

q = myQueue()
q.putall(["1", "2", "3"])
print(q.getall(2))

这将在 putall 和 getall 内发生死锁，因为两者都会尝试递归获取 mutex（首先在 putall/getall 中，然后再次put/get）。你有几个选项来解决这个问题。最简单的方法是用 threading.RLock 覆盖 Queue.__init__ 中创建的 mutex 实例。这还需要重新创建一些使用 mutex:

构建的其他 threading.Condition 对象

from Queue import Queue
import threading

class myQueue(Queue):
    def __init__(self, *args, **kwargs):
        Queue.__init__(self, *args, **kwargs)
        self.mutex = threading.RLock()
        self.not_empty = threading.Condition(self.mutex)
        self.not_full = threading.Condition(self.mutex)
        self.all_tasks_done = threading.Condition(self.mutex)

    def putall(self , objects , *args):
        with self.mutex:
            for object in objects:
                self.put(object, *args)

    def getall(self , number , *args ):
        with self.mutex:
            return [self.get(*args) for _ in xrange(number)]

q = myQueue()
q.putall(["1", "2", "3"])

print(q.getall(2))

输出：

['1', '2']

请注意，您在这里依赖 Queue 的实现细节，如果 Python 的新版本中的实现发生变化，这很容易被破坏。 myQueue 的更面向未来的版本将包装一个 Queue 实例，为其所有需要互斥锁的 public 方法提供实现，并使用您自己的递归锁来同步这些方法。

如何使 Queue 子类的 putall 和 getall 方法成为原子方法？

How can I make putall and getall methods of a Queue subclass atomic?

python

queue

multithreading