multithreading - 如何在 Python 中的循环内多线程操作

Question

假设我有一个非常大的列表，我正在执行这样的操作:

for item in items:
    try:
        api.my_operation(item)
    except:
        print 'error with item'

我的问题有两个:

• 有很多项目
• api.my_operation 需要很长时间才能返回

我想使用多线程同时启动一堆 api.my_operations，这样我就可以一次处理 5 个或 10 个甚至 100 个项目。

如果 my_operation() 返回异常(因为也许我已经处理了该项目) - 没关系。它不会破坏任何东西。循环可以继续到下一项。

注意:这是针对 Python 2.7.3

Answer 1

首先，在 Python 中，如果您的代码受 CPU 限制，那么多线程将无济于事，因为只有一个线程可以持有全局解释器锁，因此一次运行 Python 代码。所以，你需要使用进程，而不是线程。

如果您的操作“需要永远返回”则不是这样，因为它受 IO 限制，即在网络或磁盘副本等上等待。稍后我会谈到这一点。

---

接下来，一次处理 5 个或 10 个或 100 个项目的方法是创建一个由 5 个或 10 个或 100 个工作人员组成的池，并将这些项目放入工作人员服务的队列中。幸运的是，标准库 multiprocessing和 concurrent.futures库都为您包装了大部分细节。

前者对于传统编程来说更加强大和灵活；如果您需要编写future-waiting，则后者更简单；对于琐碎的情况，您选择哪个并不重要。 (在这种情况下，每个最明显的实现需要 3 行 futures，4 行 multiprocessing。)

如果您使用的是 2.6-2.7 或 3.0-3.1，则 futures 不是内置的，但您可以从 PyPI 安装它(pip install futures)。

---

最后，如果你可以将整个循环迭代变成一个函数调用(你可以这样做，例如，传递给 map)，那么并行化事情通常会简单得多，所以让我们先这样做:

def try_my_operation(item):
    try:
        api.my_operation(item)
    except:
        print('error with item')

---

把它们放在一起:

executor = concurrent.futures.ProcessPoolExecutor(10)
futures = [executor.submit(try_my_operation, item) for item in items]
concurrent.futures.wait(futures)

---

如果您有很多相对较小的作业，多处理的开销可能会淹没 yield 。解决这个问题的方法是将工作分批成更大的工作。例如(使用 itertools recipes 中的 grouper，您可以将其复制并粘贴到您的代码中，或者从 PyPI 上的 more-itertools 项目中获取):

def try_multiple_operations(items):
    for item in items:
        try:
            api.my_operation(item)
        except:
            print('error with item')

executor = concurrent.futures.ProcessPoolExecutor(10)
futures = [executor.submit(try_multiple_operations, group) 
           for group in grouper(5, items)]
concurrent.futures.wait(futures)

---

最后，如果您的代码是 IO 绑定(bind)的怎么办？然后线程与进程一样好，并且开销更少(并且限制更少，但在这种情况下，这些限制通常不会影响您)。有时，“更少的开销”足以意味着您不需要使用线程进行批处理，但您需要处理进程，这是一个不错的胜利。

那么，如何使用线程而不是进程？只需将 ProcessPoolExecutor 更改为 ThreadPoolExecutor。

如果您不确定您的代码是 CPU 密集型还是 IO 密集型，请尝试两种方式。

---

Can I do this for multiple functions in my python script? For example, if I had another for loop elsewhere in the code that I wanted to parallelize. Is it possible to do two multi threaded functions in the same script?

是的。事实上，有两种不同的方法。

首先，您可以共享同一个(线程或进程)执行程序，并在多个地方使用它没有问题。任务和 future 的全部意义在于它们是独立的。您不在乎它们在哪里运行，只需将它们排队并最终得到答案即可。

或者，您可以在同一个程序中拥有两个执行程序，这没有问题。这有性能成本——如果你同时使用两个执行器，你最终会尝试在 8 个内核上运行(例如)16 个繁忙的线程，这意味着会有一些上下文切换。但有时它是值得做的，因为，比如说，两个执行器很少同时忙，它使你的代码更简单。或者，一个执行器正在运行可能需要一段时间才能完成的非常大的任务，而另一个执行器正在运行需要尽快完成的非常小的任务，因为对于部分程序而言，响应能力比吞吐量更重要。

如果您不知道哪个适合您的程序，通常是第一个。