问题标签 [concurrent.futures]

我正在使用简单的线程模块来执行并发作业。现在我想利用并发期货模块。有人可以给我举一个使用队列和并发库的例子吗？

我收到 TypeError: 'Queue' object is not iterable 我不知道如何迭代队列

代码片段：

我想让我的类方法并行运行，但它只会产生某种我无法解决的错误。我的代码是：

如果我在 Ubuntu 机器上运行此代码，则会出现如下错误：

该方法make_readdb只是简化为示例，但它是实际代码中的瓶颈，我需要使其并行。

我正在试验 Python 3.2 中引入的新的闪亮concurrent.futures模块，我注意到，几乎使用相同的代码，使用来自 concurrent.futures 的 Pool比使用multiprocessing.Pool慢得多。

这是使用多处理的版本：

这是使用concurrent.futures：

使用从Eli Bendersky 这篇文章中提取的简单分解函数，这些是在我的计算机（i7、64 位、Arch Linux）上的结果：

我无法使用 Python 分析器分析这些，因为我遇到了 pickle 错误。有任何想法吗？

我正在尝试将从大文件的行中提取的信息发送到某个服务器上运行的进程。

为了加快速度，我想用一些并行线程来做到这一点。

使用concurrent.futures的 Python 2.7 backport我试过这个：

但是，这是有问题的，因为所有期货都会立即提交，因此我的机器内存不足，因为完整的文件被加载到内存中。

我的问题是，是否有一种简单的方法可以仅在有免费工人可用时才提交新的未来。

使用 concurrent.futures 的示例（2.7 的反向移植）：

输出：

字符串"...\_base.py", line 356, in __get_result"不是我希望看到的端点。是否可以获得引发异常的真实线路？就像是：

在这种情况下，Python3 似乎显示了正确的行号。为什么python2.7不能？有什么解决方法吗？

我是期货模块的新手，并且有一项可以从并行化中受益的任务；但我似乎无法准确地弄清楚如何为线程设置函数和为进程设置函数。我将不胜感激任何人都可以就此事提供帮助。

我正在运行粒子群优化 (PSO)。无需过多介绍 PSO 本身，以下是我的代码的基本布局：

有一个Particle类，有一个getFitness(self)方法（计算一些度量并将其存储在 中self.fitness）。PSO 模拟有多个粒子实例（很容易超过 10 个；对于某些模拟，有 100 个甚至 1000 个）。
每隔一段时间，我就必须计算粒子的适应度。目前，我在 for 循环中执行此操作：

但是，我注意到每个粒子的适应度可以相互独立地计算。这使得这种适应度计算成为并行化的主要候选者。确实，我可以做到map(lambda p: p.getFitness(args), listOfParticles)。

现在，我可以很容易地做到这一点futures.ProcessPoolExecutor：

由于调用的副作用p.getFitness存储在每个粒子本身中，我不必担心从futures.ProcessPoolExecutor().

到目前为止，一切都很好。但是现在我注意到它ProcessPoolExecutor会创建新进程，这意味着它会复制内存，这很慢。我希望能够共享内存 - 所以我应该使用线程。这很好，直到我意识到在每个进程中运行多个线程并运行多个进程可能会更快，因为多个线程仍然只在我可爱的 ​​8 核机器的一个处理器上运行。

这是我遇到麻烦的地方：
根据我看到的示例，ThreadPoolExecutor在list. 也是如此ProcessPoolExecutor。所以我不能做任何迭代的事情ProcessPoolExecutor，ThreadPoolExecutor因为那样ThreadPoolExecutor会得到一个单一的对象来处理（见我的尝试，贴在下面）。
另一方面，我不能对listOfParticles自己进行切片，因为我想ThreadPoolExecutor发挥自己的魔力来弄清楚需要多少线程。

所以，一个大问题（终于）：
我应该如何构建我的代码，以便我可以使用进程和线程有效地并行化以下内容：

这是我一直在尝试的，但我不敢尝试运行它，因为我知道它不会工作：

我会很感激任何关于如何解决这个问题的想法，甚至是关于如何改进我的方法

万一这很重要，我在 python3.3.2

我只是对我编写的一些代码感到非常困惑。我惊讶地发现：

和

产生不同的结果。第一个生成一个f返回任何类型的列表，第二个生成一个concurrent.futures.Future对象列表，然后需要使用它们的result()方法评估这些对象以获得f返回的值。

我主要担心的是这意味着executor.map不能利用concurrent.futures.as_completed，这似乎是一种非常方便的方法来评估我正在对数据库进行的一些长期运行调用的结果，因为它们变得可用。

我完全不清楚concurrent.futures.ThreadPoolExecutor对象是如何工作的——天真地，我更喜欢（有点冗长）：

更简洁executor.map，以便利用可能的性能增益。我这样做有错吗？

我正在尝试在 python 中执行多线程，并且仍在尝试了解可挑选性要求以及我可以保留多少代码 OOP。

你可以从一个类中运行 ThreadPoolExecutor 并传递一个在模块级别定义的函数吗？我的设置如下：

模块A.py：

我希望能够从一个单独的模块中调用它

模块B.py

这行得通吗？我还需要导入 foo 吗？

我正在使用 concurrent.futures.ThreadPoolExecutor 来查看是否可以从我的四核处理器（具有 8 个逻辑核心）中挤出更多的工作。所以我写了以下代码：

有趣的是，相同的功能，在 for 循环中编写时大约需要一秒钟：

...而线程池代码需要超过 10 秒。现在我知道它使用了至少 4 个线程，因为我看到每个内核上的处理器负载。但是即使使用共享内存（我可以理解为什么进程可能需要一段时间，由于内存复制），我觉得运行时的这种差异太大了。

有没有人知道为什么这可能需要这么长时间？看起来一个简单的平方运算，它确实是高度可并行化的，真的不应该花这么长时间。这可能是由于字典的数量（如果是，是什么导致那里的放缓？）？

技术细节：

• 蟒蛇 3.3.3
• 四核（8 个带超标题的逻辑核心）CPU
• MAC OSX 10.9.1（小牛队）

我正在使用 Ubuntu 12.04 服务器 x64、Python 2.7.3、futures==2.1.5、eventlet==0.14.0

有人遇到同样的问题吗？

如果您取消注释该行，此代码将挂起。我只能通过按 Ctrl+C 来停止它。在这种情况下，将打印以下 KeyboardInterrupt 回溯：https://gist.github.com/max-lobur/8526120#file-traceback

这适用于 ThreadPoolExecutor。

任何反馈表示赞赏