python - 是否可以手动锁定/解锁队列？

Question

我很好奇是否有办法手动锁定 multiprocessing.Queue 对象。

我有一个非常标准的生产者/消费者模式设置，其中我的主线程不断产生一系列值，multiprocessing.Process 工作池对产生的值进行操作。

它全部由一个单独的 multiprocessing.Queue() 控制。

import time
import multiprocessing


class Reader(multiprocessing.Process): 
    def __init__(self, queue): 
        multiprocessing.Process.__init__(self)
        self.queue = queue 

    def run(self): 
        while True: 
            item = self.queue.get() 
            if isinstance(item, str): 
                break 


if __name__ == '__main__': 

    queue = multiprocessing.Queue()
    reader = Reader(queue)
    reader.start()

    
    start_time = time.time()
    while time.time() - start_time < 10: 
        queue.put(1)
    queue.put('bla bla bla sentinal')
    queue.join() 

我遇到的问题是，我的工作线程池无法像主线程向其中插入值那样快地消耗和处理 queue。因此，一段时间后，Queue 变得笨拙，以至于弹出 MemoryError。

一个明显的解决方案是简单地在生产者中添加一个等待检查，以阻止它向队列中放入更多的值。类似的东西:

while time.time() - start_time < 10: 
    queue.put(1)
    while queue.qsize() > some_size:
        time.sleep(.1)
queue.put('bla bla bla sentinal')
queue.join() 

但是，由于该程序的时髦性质，我想将队列中的所有内容转储到文件中以供以后处理。但!如果不能暂时锁定队列，工作人员就无法消费队列中的所有内容，因为生产者不断地用垃圾填满它——无论如何在概念上都是如此。经过无数次测试后，似乎在某个时候其中一把锁获胜(但通常是添加到队列中的那把锁)。

编辑:此外，我意识到可以简单地停止生产者并从该线程使用它......但这让我内心的单一责任人感到难过，因为生产者是生产者，而不是消费者.

编辑:

在查看了Queue 的源码后，我想到了这个:

def dump_queue(q):
    q._rlock.acquire()
    try:
        res = []
        while not q.empty():
            res.append(q._recv())
            q._sem.release()
        return res
    finally:
        q._rlock.release()    

但是，我太害怕使用它了!我不知道这是否“正确”。我没有足够的把握知道这是否会在不破坏任何 Queue 内部结构的情况下保持下去。

有人知道这会不会坏吗？ :)

Answer 1

鉴于评论中所说的内容，Queue 只是针对您的问题的错误数据结构 - 但很可能是可用解决方案的一部分。

听起来您只有一个制作人。创建一个新的本地生产者(不跨进程共享)类来实现您真正需要的语义。例如，

class FlushingQueue:
    def __init__(self, mpqueue, path_to_spill_file, maxsize=1000, dumpsize=1000000):
        from collections import deque
        self.q = mpqueue  # a shared `multiprocessing.Queue`
        self.dump_path = path_to_spill_file
        self.maxsize = maxsize
        self.dumpsize = dumpsize
        self.d = deque()  # buffer for overflowing values

    def put(self, item):
        if self.q.qsize() < self.maxsize:
            self.q.put(item)
            # in case consumers have made real progress
            while self.d and self.q.qsize() < self.maxsize:
                self.q.put(self.d.popleft())
        else:
            self.d.append(item)
            if len(self.d) >= self.dumpsize:
                self.dump()

    def dump(self):
        # code to flush self.d to the spill file; no
        # need to look at self.q at all

我敢打赌你可以完成这项工作:-)