For循环上的Python多处理

Posted 2023-02-16

【中文标题】For循环上的Python多处理

【英文标题】：Python multiprocessing on For Loop

【发布时间】：2016-03-31 13:10:39

【问题描述】：

首先，我知道在 python 上已经有很多关于多处理的线程，但是这些似乎都不能解决我的问题。

这是我的问题： 我想实现随机森林算法，一个天真的方法是这样的：

def random_tree(Data):
    tree = calculation(Data)
    forest.append(tree)

forest = list()
for i in range(300):
    random_tree(Data)

里面有 300 棵“树”的forest 将是我的最终结果。在这种情况下，如何将此代码转换为多处理版本？

---

更新： 我刚刚在一个非常简化的脚本中尝试了 Mukund M K 的方法：

from multiprocessing import Pool

def f(x):
    return 2*x

data = np.array([1,2,5])

pool = Pool(processes=4)
forest = pool.map(f, (data for i in range(4))) 
# I use range() instead of xrange() because I am using Python 3.4

现在......脚本一直在运行......我打开一个python shell并逐行输入脚本，这是我得到的消息：

> 进程 SpawnPoolWorker-1： > 进程 SpawnPoolWorker-2： > Traceback（最近一次通话最后一次）： > 处理 SpawnPoolWorker-3： > Traceback（最近一次通话最后一次）： > 进程 SpawnPoolWorker-4： > Traceback（最近一次通话最后一次）： > Traceback（最近一次通话最后一次）： > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 254 行，在 _bootstrap 自我运行（） > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 254 行，在 _bootstrap 自我运行（） > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 254 行，在 _bootstrap 自我运行（） > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 254 行，在 _bootstrap 自我运行（） > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) > 文件“E:\Anaconda3\lib\multiprocessing\process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) > 文件“E:\Anaconda3\lib\multiprocessing\pool.py”，第 108 行，在 worker 中 任务 = 获取（） > 文件“E:\Anaconda3\lib\multiprocessing\pool.py”，第 108 行，在 worker 中 任务 = 获取（） > 文件“E:\Anaconda3\lib\multiprocessing\pool.py”，第 108 行，在 worker 中 任务 = 获取（） > 文件“E:\Anaconda3\lib\multiprocessing\pool.py”，第 108 行，在 worker 中 任务 = 获取（） > 文件“E:\Anaconda3\lib\multiprocessing\queues.py”，第 357 行，在 get 返回 ForkingPickler.loads(res) > 文件“E:\Anaconda3\lib\multiprocessing\queues.py”，第 357 行，在 get 返回 ForkingPickler.loads(res) > AttributeError: 无法获取属性“f” > AttributeError: 无法获取属性“f” 文件“E:\Anaconda3\lib\multiprocessing\queues.py”，第 357 行，在 get 返回 ForkingPickler.loads(res) > AttributeError: 无法获取属性“f” 文件“E:\Anaconda3\lib\multiprocessing\queues.py”，第 357 行，在 get 返回 ForkingPickler.loads(res) > AttributeError: 无法获取属性“f”

---

更新：我根据其他一些示例代码编辑了我的示例代码，如下所示：

from multiprocessing import Pool
import numpy as np

def f(x):
    return 2*x

if __name__ == '__main__':
    data = np.array([1,2,3])
    with Pool(5) as p:
        result = p.map(f, (data for i in range(300)))

现在可以了。我现在需要做的是现在用更复杂的算法来填充它.. 我脑海中的另一个问题是：为什么这段代码可以工作，而以前的版本不能？

【问题讨论】：

“数据”是一个二维(100*3) numpy 数组。

你是在阅读还是在计算中修改内容？如果是，修改顺序是否重要？

我只读取数据。在随机森林算法中，我会从原始数据（“数据”）中随机抽样来构建一棵树。所以每次迭代都是独立的，这就是为什么我认为它应该能够并行化。

我知道这是旧的，但以防万一。这里的罪魁祸首可能是失踪的if __name__ == '__main__':。如果您阅读多处理 python 文档，您会发现这是 mp 工作的明确要求。

【参考方案1】：

您可以通过这种方式进行多处理：

from multiprocessing import Pool

def random_tree(Data):
    return calculation(Data)

pool = Pool(processes=4)
forest = pool.map(random_tree, (Data for i in range(300)))

【讨论】：

我不明白这个建议的重点。但是我在我的测试脚本中尝试过（定义函数后初始化池），它仍然运行并且没有停止。

好吧，当我在 python3 上运行它时，它似乎腌制了数据。由于该函数是稍后定义的，因此无法找到它并引发错误。该代码似乎对我有用。您的简化脚本也可以工作。

试图在 python shell 中运行脚本，但仍然不起作用。（见上文）

this 可能有帮助吗？由于您使用的是Windows。

不，这并没有真正帮助我解决我原来的问题。但不知何故，我找到了一种让我的示例代码工作的方法。（查看更新）

【参考方案2】：

包裹处理可能会对您有所帮助。看看here。