Python使用队列列表进行多进程处理

Question

我尝试使用Python 2.7的多进程包创建队列列表。每个子进程拥有一个独立的队列，并且有两个任务：从自己的队列中获取元素并将元素放入其他子进程的队列中。因此，每个子进程都必须知道它所属的队列，这就是我使用队列列表的原因。

我的代码如下：

mgr = multiprocessing.Manager()

sharedQueueList = mgr.list()

for i in xrange(num_processes):
  sharedQueueList .append(mgr.Queue())

但是，我收到了以下错误信息：

**raise convert_to_error(kind, result)**
RemoteError: 
---------------------------------------------------------------------------
Unserializable message: ('#RETURN', < Queue.Queue instance at 0x02AD3170 >)
---------------------------------------------------------------------------

- jmguo

Answer 1

在父进程中创建Queue列表，并在创建时将其中几个分配给每个工作进程。每个工作进程将从它的一个队列中获取任务，然后输出到另一个队列中。

import logging, multiprocessing

def myproc(arg):
    return arg*2

def worker(qlist):
    logger = multiprocessing.get_logger()
    logger.info('start')
    while True:
        job = qlist[0].get()
        logger.info('got %s', job)
        if job is None:
            logger.info('exiting')
            return
        qlist[1].put( myproc(job) )

logger = multiprocessing.log_to_stderr(
    level=logging.INFO,
)
logger.info('setup')

numProcs = 3
queueList = [ multiprocessing.Queue() for i in xrange(numProcs) ]

# prefill with 3 jobs
for num in range(3):
    queueList[0].put(num)
# signal end of jobs
queueList[0].put(None)

worker_p = multiprocessing.Process(
    target=worker, args=( [queueList[0], queueList[1]], ),
    name='worker',
)
worker_p.start()

worker_p.join()

logger.info('done')

示例运行：

[INFO/MainProcess] setup
[INFO/worker] child process calling self.run()
[INFO/worker] start
[INFO/worker] got 0
[INFO/worker] got 1
[INFO/worker] got 2
[INFO/worker] got None
[INFO/worker] exiting
[INFO/worker] process shutting down
[INFO/worker] process exiting with exitcode 0
[INFO/MainProcess] done
[INFO/MainProcess] process shutting down

- johntellsall