相关疑难解决方法(0)

我知道multiprocessing.Manager()它是如何用于创建共享对象的,特别是可以在worker之间共享的队列.有这个问题,这个问题,这个问题甚至是我自己的一个问题.

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

当我需要放置和获取数据时,我想使用字典来访问我的队列.我无法做到这一点.我尝试过很多东西.随着multiprocessing进口为mp:

for key in all_keys: DICT[key] = mp.Queue在多处理模块导入的配置文件中定义一个dict (调用它multi.py)不会返回错误,但是队列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-加载第一个模块,它定义一些变量,导入multi,启动multi.main()和加载另一个模块,该模块启动一系列模块加载和代码执行.与此同时...

2- …

我有一个下载器功能,可以并行下载多个文件.我用multiprocessing.Pool.map_async它来下载同一文件的不同块.我想显示下载的状态栏.为此,我需要知道已经下载的总字节数(total_bytes_dl).

    pool = multiprocessing.Pool(processes)
    mapObj = pool.map_async(f, args)

    while not mapObj.ready():
        status = r"%.2f MB / %.2f MB" % (total_bytes_dl / 1024.0 / 1024.0, filesize / 1024.0 / 1024.0,)
        status = status + chr(8)*(len(status)+1)
        print status,
        time.sleep(0.5)

有没有办法设置一个将在所有这些进程和主进程之间共享的变量,因此每个进程都可以附加刚刚下载的字节数量？

我正在使用 ProcessPoolExecutor 上下文管理器并行运行多个 Kafka 使用者。我需要存储子进程的进程 ID，以便稍后我可以干净地终止这些进程。我有这样的代码：

Class MultiProcessConsumer:
    ...

    def run_in_parallel(self):
        parallelism_factor = 5

        with ProcessPoolExecutor() as executor:
            processes = [executor.submit(self.consume) for _ in range(parallelism_factor)]
            # It would be nice If I could write [process.pid for process in processes] to a file here.

    def consume(self):
        while True:
            for message in self.kafka_consumer:
                do_stuff(message)

os.get_pid()我知道我可以在 Consumer 方法中使用来获取 PID。但是，正确处理它们（在消费者不断关闭或启动的情况下）需要一些额外的工作。

您建议我如何在这样的上下文中获取并存储子进程的 PID？

我想在多个进程之间共享numpy数组.有工作的解决方案在这里.但是它们都通过继承将数组传递给子进程,这对我来说不起作用,因为我必须事先启动一些工作进程,而且我不知道稍后我要处理多少个数组.有没有办法在进程启动后创建这样的数组并通过队列将这些数组传递给进程？

顺便说一下,我无法使用multiprocessing.Manager.

我知道 multiprocessing.Manager() 以及如何使用它来创建共享对象。特别是可以在工作人员之间共享的队列。有这个问题，这个问题，还有这个问题。

然而，这些链接并没有提到为什么我们可以使用继承来在进程之间进行共享。据我了解，在这种情况下仍然只能复制队列。

相关问题出现在Why I can't use multiprocessing.Queue with ProcessPoolExecutor? 。我提供了部分答案以及解决方法，但承认这个问题引发了另一个问题，即为什么可以multiprocessing.Queue将实例作为参数传递给辅助函数。multiprocessing.Process

from multiprocessing import Pool, Queue

def worker(q):
    print(q.get())

with Pool(1) as pool:
    q = Queue()
    q.put(7)
    pool.apply(worker, args=(q,))

from multiprocessing import Process, Queue

def worker(q):
    print(q.get())

q = Queue()
q.put(7)
p = Process(target=worker, args=(q,))
p.start()
p.join()