如何使用线程并行压缩迭代器？答案

【问题标题】：How to zip iterators in parallel using threading?如何使用线程并行压缩迭代器？
【发布时间】：2019-02-20 13:41:23
【问题描述】：

假设我有 N 个生成器，可以生成 gs = [..] # list of generators 的项目流。

我可以很容易地将zip 他们一起从gs:tuple_gen = zip(*gs) 中的每个生成器中获取一个元组生成器。

这会在gs 中按顺序对每个g 调用next(g)，并将结果收集到一个元组中。但是，如果每个项目的生产成本都很高，我们可能希望在多个线程上并行处理 next(g) 的工作。

我怎样才能实现一个 pzip(..) 来做到这一点？

【问题讨论】：

a) 哪个操作系统？ b) N 有多大？ c) 为每个生成器生产一个项目的时间是否相当相等？ d) 你知道GIL 吗？如果您的生成器涉及 Python 中的 cpu-bound 工作（不是 I/0 或使用像 numpy 这样的释放 GIL 的 C 扩展），您将需要真正并行执行的进程。
1. Ubuntu。 2. N~32 3. 是的。 4. IO绑定读取文件。
如果您受磁盘访问的限制，那么更多线程可能不会帮助您，因为在大多数情况下限制是原始磁盘访问速度。但是，如果您受 IO 限制并正在等待（即用于用户输入或慢速网络连接），则运行多个线程可能非常有益。
相关：使用进程的真正并行生成器覆盖here。

标签： python multithreading parallel-processing iterator generator

【解决方案1】：

您所要求的可以通过创建一个生成器来实现，该生成器从 ThreadPool 上的apply_async-calls 中产生结果。

仅供参考，我使用 pandas.read_csv-iterators 对这种方法进行了基准测试，您可以通过指定 chunksize 参数获得。我创建了 1M 行大小的 csv 文件的 8 个副本，并指定了 chunksize=100_000。

其中四个文件是使用您提供的顺序方法读取的，四个使用下面的mt_gen 函数，使用四个线程池：

单线程 ~ 3.68 秒

多线程 ~ 1.21 秒

但这并不意味着它会改善每个硬件和数据设置的结果。

import time
import threading
from multiprocessing.dummy import Pool  # dummy uses threads


def _load_sim(x = 10e6):
    for _ in range(int(x)):
        x -= 1
    time.sleep(1)


def gen(start, stop):
    for i in range(start, stop):
        _load_sim()
        print(f'{threading.current_thread().name} yielding {i}')
        yield i


def multi_threaded(gens):
    combi_g = mt_gen(gens)
    for item in combi_g:
        print(item)


def mt_gen(gens):
    with Pool(N_WORKERS) as pool:
        while True:
            async_results = [pool.apply_async(next, args=(g,)) for g in gens]
            try:
                results = [r.get() for r in async_results]
            except StopIteration:  # needed for Python 3.7+, PEP 479, bpo-32670
                return
            yield results


if __name__ == '__main__':

    N_GENS = 10
    N_WORKERS = 4
    GEN_LENGTH = 3

    gens = [gen(x * GEN_LENGTH, (x + 1) * GEN_LENGTH) for x in range(N_GENS)]
    multi_threaded(gens)

输出：

Thread-1 yielding 0
Thread-2 yielding 3
Thread-4 yielding 6
Thread-3 yielding 9
Thread-1 yielding 12
Thread-2 yielding 15
Thread-4 yielding 18
Thread-3 yielding 21
Thread-1 yielding 24
Thread-2 yielding 27
[0, 3, 6, 9, 12, 15, 18, 21, 24, 27]
Thread-3 yielding 7
Thread-1 yielding 10
Thread-2 yielding 4
Thread-4 yielding 1
Thread-3 yielding 13
Thread-1 yielding 16
Thread-4 yielding 22
Thread-2 yielding 19
Thread-3 yielding 25
Thread-1 yielding 28
[1, 4, 7, 10, 13, 16, 19, 22, 25, 28]
Thread-1 yielding 8
Thread-4 yielding 2
Thread-3 yielding 11
Thread-2 yielding 5
Thread-1 yielding 14
Thread-4 yielding 17
Thread-3 yielding 20
Thread-2 yielding 23
Thread-1 yielding 26
Thread-4 yielding 29
[2, 5, 8, 11, 14, 17, 20, 23, 26, 29]

Process finished with exit code 0

【讨论】：