Python multiprocessing que is duplicated

Question

我正在创建子进程作为worker对象，这些对象通过input和output队列获取和返回数据。但是，当代码运行时，似乎每个子进程都获取了input队列的完整副本，而主线程没有通过队列获取output任何内容，从而导致死锁。为什么？

import threading, queue, multiprocessing
class worker(multiprocessing.Process):
     def __init__(self,inp,out):
         super().__init__()
         self.input=inp
         self.output=out
         #real program has external subprocess initialization here
     def run(self):
         name=self.name
         while True:
             inp=self.input.get()
             if (inp is 'stop'):
                 break
             print('Process {} got task word: {}'.format(name,inp))
             print('queue size is: {}'.format(self.input.qsize()))
             self.output.put(inp.count('a'))
             print('Process {} input processed'.format(name))
         print('exiting {}'.format(name))
if __name__ == "__main__":
    inp=queue.Queue()
    out=queue.Queue(maxsize=4)
    strings = ['asd', 'assa','aaa','as','aa','aaaq']
    for x in strings:
        inp.put(x)
    print(inp.qsize())
    workers=[worker(inp,out) for x in range(2)]
    for w in workers:
        w.start()
    res=[]
    for x in strings:
        res.append(out.get())
        print("intermediate result is {}".format(res))
    for _ in workers:
        inp.put('stop')
    for w in workers:
        w.join()
    print(res)

在实际问题中，每个工作线程都会将自己的接口初始化到另一个外部程序。我想避免为每个项目重复该初始化，因此我不能简单地将 worker 转换为函数并将其与 map 一起使用。

如果multiprocessing.Process替换为具有相同 API 的 threading.Thread，则程序可以正常工作，但在单核上，这对于任务来说是不可接受的。

Answer 1

不同的进程不共享进程内存，因此您无法在不同进程之间共享队列。每个进程都有自己的副本。

Python多个进程是否共享同一对象？

您必须使用 IPC 机制来同步多个进程。

但是，线程

共享进程内存，因此它适用于线程，但不适用于进程。

多进程解决方案

使用多处理队列

import threading, multiprocessing
class worker(multiprocessing.Process):
     def __init__(self,inp,out):
         super().__init__()
         self.input=inp
         self.output=out
         #real program has external subprocess initialization here
     def run(self):
         name=self.name
         while True:
             inp=self.input.get()
             if (inp == 'stop'):
                 break
             print('Process {} got task word: {}'.format(name,inp))
             #print('queue size is: {}'.format(self.input.qsize()))
             self.output.put((inp, inp.count('a')))
             print('Process {} input processed'.format(name))
         print('exiting {}'.format(name))
if __name__ == "__main__":
    inp=multiprocessing.Queue()
    out=multiprocessing.Queue()
    strings = ['asd', 'assa','aaa','as','aa','aaaq']
    for x in strings:
        inp.put(x)
    workers=[worker(inp,out) for x in range(2)]
    for w in workers:
        w.start()
    for _ in workers:
        inp.put('stop')
    #print (inp.qsize())

    inp.close()
    inp.join_thread()
    for w in workers:
        w.join()
    print (out.empty())
    while not out.empty():
        print (out.get())