问题标签 [multiprocessing-manager]

我想从网站下载和处理很多文件。该网站的服务条款限制了您每秒允许下载的文件数量。

处理文件所需的时间实际上是瓶颈，所以我希望能够并行处理多个文件。但我不希望不同的进程结合起来违反下载限制。所以我需要一些限制过度请求率的东西。我在想类似以下的事情，但我并不完全是该multiprocessing模块的专家。

然后在其他地方运行下载

这似乎在小范围内完成了这项工作，但我有点担心锁定是否真的正确完成。

此外，如果有更好的模式来实现相同的目标，我很想听听。

考虑以下python代码：

在此示例中，我创建了一个托管 dict，其中包含一个列表作为MyClass. 目标是在多处理环境中增加此列表的某些元素，但某些方法不能有效地修改列表。

方法1： dictSumOverloaded
重载运算符+=用于将列表的元素加1，但结果不持久。字典没有更新。
方法 2： dictSumElementWise
此函数基于旧列表和要添加的值明智地创建一个新的列表元素。然后将新列表分配给 dict 键。dict 已成功修改。
健全性检查：在多处理环境之外
dic3使用时，在多处理环境之外按预期进行修改+=。

问题：
1）为什么+=不在多处理环境中修改列表元素？
2）使用元素明智的方法来更新列表工作但很麻烦，有什么建议让它更干净/更快？

所以我试图在一个字典上使用多处理管理器，这是我最初的尝试：

当然，当我运行它时，输出不是我所期望的，z正确更新而y不变！这是输出：

然后我用谷歌搜索，在这里找到了一个解释，显然嵌套的字典也必须是Manager().dict()s 而不是普通的 python 字典（可能从 Python 3.6 开始）。所以我做了以下事情：

但是，我得到了这个无法解释的错误，而不是它正常工作，为了清楚起见，它分为三个部分。第一部分对应于，test['y']['Y0'] += 5而第二部分是简单的print(test)，最后是输出print(test['y'])

我不确定为什么会这样。内部 dicts 显然是被创建的（如输出的第二部分所示）。但是由于某种原因，它们根本无法读取或写入！为什么会这样？

额外：如果我通过 python 控制台（而不是脚本）运行相同的 python 代码，错误会FileNotFoundError从ConnectionRefusedError. 但是具有相同的确切回溯！

我的主要目标是从父进程中打开 30 个子进程，然后从这 30 个子进程中的每一个中打开未知数量的新进程。我将调用 redis 以获取来自那些新子进程的一些位置数据，我不确定我必须调用多少次，它会是 100 次或超过 1000 次。当我调用超过 1000 次时，我正在跨页限制，错误是：

我不想手动增加生产服务器上的页面限制。我想限制进程的创建，这样它就不会打开超过 1000 个连接。

这是我的模板代码：

谢谢你。

如SO问题中所述：

我可以在多处理池星图方法中传递队列对象吗

在多个进程之间共享一个结果队列

当使用多处理的 Pool 方法时，multiprocessing.Queue() 不起作用并在将队列传递给需要处理的方法时出错。但是当使用 Manager.Queue() 时，它工作正常。没有解释为什么这在任何答案中都有效，所以想为此提出一个单独的问题。

如上所述，其他看似重复的问题通过使用 manager.Queue 提供了答案。但是他们都没有解释为什么这会有所帮助；我已经问过这个问题，以便解释为什么这种方法有效而 multiprocessing.Queue() 无效。

我对 python 多处理完全陌生，对大量的在线资源有点不知所措，所以我有点想从这里获得更清晰的方法。我的代码如下所示：向前和向后这两个函数在计算上非常昂贵。在我的输入数据集上，每个大约需要 13 分钟。我想同时计算两个矩阵（向前和向后，参见 decode() 函数中的第 3 行和第 4 行代码）。我查看了一些在线教程，我想我可以使用 multiprocessing.process 来做到这一点。但是，我不确定如何检索 numpy 数组。我知道有诸如 Queue、Array 之类的东西，但它们的使用似乎非常有限，而且似乎不适合这里。提前致谢！'''

'''

After my previous thread was marked as a duplicate, it pointed me in the direction of multiprocessing managers. I'm trying to use multiprocessing to create a service that handles my pandas dataframe to give to Flask requests. This is my code so far:

df_manager.py

data_handler.py

Unfortunately, this throws an exception when attempting to call manager.get_df() in data_handler.py.

Any help in the right direction would be much appreciated!

EDIT: This seems to be caused by DataFrames specifically, as returning df.to_json() instead of just df in df_manager.py seems to work fine. Still investigating...

EDIT2: I have updated the code to remove the Flask dependency, as it has seemingly nothing to do with it.

Git repo

我有一个启动多个进程的 python 程序。而这些进程又会产生它们自己的子进程（嵌套进程）。我喜欢保留由主进程或任何子进程启动的所有进程的列表，但无法弄清楚如何将嵌套进程的句柄返回给主进程。这是我尝试过的，但编译器说进程句柄不可选择。

我正在使用multiprocessingpython 3.7 中的库。

在我的用例中，我从正常list的dicts 开始 - 我使用它从 a 创建一个ListProxy对象，SyncManager然后在池中使用它来同步list（创建如下）。

ListProxy我的问题：一旦池完成运行，是否有一种 pythonic 方法可以将我的背部更改为正常列表？我查看了文档，甚至查看了源代码，似乎唯一的方法可能是使用类中的_getvalue()方法BaseProxy- 但这似乎很老套，而且我还必须递归地“取消代理”任何嵌套的代理对象。

我需要返回正常列表，以便稍后在代码中将其序列化为 json，ProxyObjects 不容易序列化。

我正在尝试使用 python 多处理模块在多个设备上并行运行一堆 linux 命令。脚本能够串行连接到多个设备并开始并行执行 linux 命令。该脚本在第一次尝试中失败，但在第二次尝试中成功。

请查找错误日志：

代码：

任何人都可以在这里帮忙。