具有 SyncManager 类的 python 中多进程优先级队列的队列函数答案

【问题标题】：queue function for Multiprocess Priority Queue in python with SyncManager class具有 SyncManager 类的 python 中多进程优先级队列的队列函数
【发布时间】：2018-12-04 13:00:57
【问题描述】：

我想实现多处理优先队列。我找到了这个答案：- Strange Queue.PriorityQueue behaviour with multiprocessing in Python 2.7.6

Dano

在我实现了这个之后。我可以为我的优先队列使用 .get() 和 .put() 函数，但是当我使用 .queue 打印队列中的当前元素时，它给了我一个错误

代码：-

 class MyManager(SyncManager):   
    pass

def get_manager():
    MyManager.register("PriorityQueue", PriorityQueue)  # Register a shared PriorityQueue
    m = MyManager()
    m.start()
    return m

m = get_manager()
call= m.PriorityQueue()
for i in range(5):
    call.put(i)

print(call.queue)

错误：AttributeError: 'AutoProxy[PriorityQueue]' object has no attribute 'queue'

我阅读了 SyncManager 的 python 文档并修改了我的代码。

新代码：-

class MyManager(SyncManager):   
    pass

def get_manager():
    MyManager.register("PriorityQueue", PriorityQueue,exposed=['put','get','queue'])  # Register a shared PriorityQueue
    m = MyManager()
    m.start()
    return m

m = get_manager()
call= m.PriorityQueue()
for i in range(5):
    call.put(i)

print(call.queue)

现在的输出是：-

<bound method AutoProxy[PriorityQueue].queue of <AutoProxy[PriorityQueue] object, typeid 'PriorityQueue' at 0x7ff3b48f2dd0>>

我仍然没有得到队列中的元素，我阅读了关于注册函数的method_to_typeid 属性来映射exposed 中提到的函数的返回类型，但我不知道如何使用它。

有人可以帮我解决这个问题，这样我就可以打印队列的元素而不用从队列中弹出它们

【问题讨论】：

queue 似乎是一个方法，所以你需要调用它。试试call.queue()（如果它需要任何参数，它可能会给你另一个错误）。
queue 不可调用。所以我们不需要括号 () 和 queue ，无论是普通队列、多处理队列还是任何其他类型的队列

标签： python multiprocessing priority-queue

【解决方案1】：

您只能通过代理使用所指对象的方法。由于PriorityQueue().queue 不是方法，而是实例属性，因此您需要提供一个可以返回该属性值的方法。下面的示例选择了具有子类化PriorityQueue 的通用get_attribute 方法。

# Python 3.7.1
from queue import PriorityQueue
from multiprocessing.managers import SyncManager
from multiprocessing import Process


SENTINEL = None


class MyPriorityQueue(PriorityQueue):
    def get_attribute(self, name):
        return getattr(self, name)


class MyManager(SyncManager):
    pass


def get_manager():
    MyManager.register("PriorityQueue", MyPriorityQueue)
    m = MyManager()
    m.start()
    return m


def f(q):
    for item in iter(lambda: q.get()[1], SENTINEL):
        print(item)
    print(f'queue: {q.get_attribute("queue")}')


if __name__ == '__main__':

    m = get_manager()
    pq = m.PriorityQueue()

    tasks = enumerate([f'item_{i}' for i in range(5)] + [SENTINEL])

    for task in tasks:
        pq.put(task)

    print(f'queue: {pq.get_attribute("queue")}')
    print(f'maxsize: {pq.get_attribute("maxsize")}')

    p = Process(target=f, args=(pq,))
    p.start()
    p.join()

示例输出：

queue: [(0, 'item_0'), (1, 'item_1'), (2, 'item_2'), (3, 'item_3'), (4, 'item_4'), (5, None)]
maxsize: 0
item_0
item_1
item_2
item_3
item_4
queue: []

【讨论】：

谢谢，工作完美，也消除了我的疑虑。