在Python中的进程之间共享许多队列

Question

我知道multiprocessing.Manager()如何将其用于创建共享对象，尤其是可以在工作人员之间共享的队列。有这个问题，这个问题，[这个问题](http://codingdict.com/questions/1299甚至是我自己的一个问题。

但是，我需要定义很多队列，每个队列都链接一对特定的进程。假设每对进程及其链接队列均由变量标识key。

当我需要放置和获取数据时，我想使用字典来访问我的队列。我无法完成这项工作。我已经尝试了很多东西。随着multiprocessing进口为mp：

像for key in all_keys: DICT[key] = mp.Queue在多处理模块（称为multi.py）导入的配置文件中那样定义dict不会返回错误，但是DICT[key]在进程之间不共享队列，每个进程似乎都有自己的队列副本，因此不会发生通信。

如果我尝试在定义DICT进程并启动它们的主要多处理函数的开始处定义，例如

DICT = mp.Manager().dict()    
for key in all_keys:
    DICT[key] = mp.Queue()

我得到错误

RuntimeError: Queue objects should only be shared between processes through
 inheritance

更改为

DICT = mp.Manager().dict()    
for key in all_keys:
    DICT[key] = mp.Manager().Queue()

只会使一切变得更糟。multi.py在main函数的顶部而不是内部尝试类似的定义会返回类似的错误。

必须有一种方法可以在进程之间共享许多队列，而无需在代码中明确命名每个队列。有任何想法吗？

编辑

这是程序的基本架构：

1-加载第一个模块，该模块定义一些变量，import
multi，launchsmulti.main()和加载另一个模块，该模块开始一系列模块加载和代码执行。与此同时…

2-multi.main看起来像这样：

def main():
    manager = mp.Manager()
    pool = mp.Pool()
    DICT2 = manager.dict()

    for key in all_keys:
        DICT2[key] = manager.Queue()
        proc_1 = pool.apply_async(targ1,(DICT1[key],) ) #DICT1 is defined in the config file
        proc_2 =  pool.apply_async(targ2,(DICT2[key], otherargs,)

除了使用pooland之外manager，我还启动了以下程序：

mp.Process(target=targ1, args=(DICT[key],))

3-该函数targ1接收key来自主进程的输入数据（按排序）。它旨在将结果传递给它，DICT[key]以便targ2可以执行其工作。这是不起作用的部分。有任意数量的targ1s，targ2s等，因此有任意数量的队列。

4-这些过程中的某些过程的结果将被发送到一堆不同的数组/熊猫数据帧，这些数据帧也由索引key，我希望可以从任意过程中访问它们，甚至是在不同模块中启动的过程。我还没有写这部分，这可能是一个不同的问题。（我在这里提到它是因为上面3的答案也可能很好地解决了4。）

Answer 1

当您尝试multiprocessing.Queue()通过传递a作为参数共享a时，听起来好像问题开始了。您可以通过创建托管队列来解决此问题：

import multiprocessing
manager = multiprocessing.Manager()
passable_queue = manager.Queue()

当使用管理器创建它时，您正在存储并传递一个 代理
到队列，而不是队列本身，因此即使传递给工作进程的对象是复制的，它仍将指向相同的基础数据结构：您的队列。在概念上，它与C / C
++中的指针非常相似。如果以这种方式创建队列，则在启动工作进程时将能够通过它们。

由于您现在可以传递队列，因此不再需要管理字典。在main中保留一个普通字典，该字典将存储所有映射，并且仅为您的工作进程提供所需的队列，因此他们无需访问任何映射。

我在这里写了一个例子。看起来您在工作人员之间传递对象，这就是在这里完成的工作。假设我们有两个处理阶段，并且数据在的控制下开始和结束main。看看我们如何创建像管道一样连接工人的队列，但是通过
只 给他们提供 他们需要的队列 ，就不需要他们知道任何映射：

import multiprocessing as mp

def stage1(q_in, q_out):

    q_out.put(q_in.get()+"Stage 1 did some work.\n")
    return

def stage2(q_in, q_out):

    q_out.put(q_in.get()+"Stage 2 did some work.\n")
    return

def main():

    pool = mp.Pool()
    manager = mp.Manager()

    # create managed queues
    q_main_to_s1 = manager.Queue()
    q_s1_to_s2 = manager.Queue()
    q_s2_to_main = manager.Queue()

    # launch workers, passing them the queues they need
    results_s1 = pool.apply_async(stage1, (q_main_to_s1, q_s1_to_s2))
    results_s2 = pool.apply_async(stage2, (q_s1_to_s2, q_s2_to_main))

    # Send a message into the pipeline
    q_main_to_s1.put("Main started the job.\n")

    # Wait for work to complete
    print(q_s2_to_main.get()+"Main finished the job.")

    pool.close()
    pool.join()

    return

if __name__ == "__main__":
    main()

代码产生以下输出：

Main开始了工作。
第一阶段做了一些工作。
第二阶段做了一些工作。
Main完成了工作。

我没有提供AsyncResults在字典中存储队列或对象的示例，因为我仍然不太了解您的程序应该如何工作。但是，既然您可以自由地传递队列，则可以构建字典来根据需要存储队列/进程映射。

实际上，如果您确实在多个工作人员之间建立了一条管道，那么您甚至不需要保留对中“工作人员之间”队列的引用main。创建队列，将其传递给您的工作人员，然后仅保留对main将要使用的队列的引用。如果您确实有“任意数量”的队列，我绝对建议您尝试尽快让旧队列被垃圾回收。