Python 多个进程消耗/迭代单个生成器（分而治之）答案

【问题标题】：Python multiple processes consuming/iterating over single generator (divide and conquer)Python 多个进程消耗/迭代单个生成器（分而治之）
【发布时间】：2019-08-06 15:27:05
【问题描述】：

我有一个返回大量项目的 python 生成器，例如：

import itertools

def generate_random_strings():
    chars = "ABCDEFGH"
    for item in itertools.product(chars, repeat=10):
        yield "".join(item)

然后我对此进行迭代并执行各种任务，问题是我只为此使用一个线程/进程：

my_strings = generate_random_strings()
for string in my_strings:
    # do something with string...
    print(string)

这很好用，我得到了所有的字符串，但是速度很慢。我想利用 Python 多处理的力量来“分而治之”这个 for 循环。但是，当然，我希望每个字符串只处理一次。虽然我找到了很多关于多处理的文档，但我正在尝试用最少的代码找到最简单的解决方案。我假设每个线程每次都应该获取一大块项目并在返回并获取另一个大块之前处理它们......

非常感谢，

【问题讨论】：

关于速度的旁注：它不会有很大的不同，但是如果你将for item in itertools.product(chars, repeat=10): yield "".join(item) 更改为yield from map( "".join, itertools.product(chars, repeat=10))，它会产生更快的结果（通过将所有工作推到 C 层，消除字节码解释器开销，并确保在请求下一个之前释放product 的tuples，这允许它重用tuple 而不是分配新的）。
感谢您的提示！

标签： python python-3.x multiprocessing generator python-multiprocessing

【解决方案1】：

代码最少的最简单解决方案？多处理上下文管理器。

我假设您可以将“用字符串做某事”放入一个名为“do_something”的函数中

from multiprocessing import Pool as ProcessPool

number_of_processes = 4

with ProcessPool(number_of_processes) as pool:
    pool.map(do_something, my_strings)

如果您想再次获得“do_something”的结果，很简单！

with ProcessPool(number_of_processes) as pool:
    results = pool.map(do_something, my_strings)

您会在列表中找到它们。

Multiprocessing.dummy 是进程池的语法包装器，可让您使用多处理语法。如果你想要线程而不是进程，只需这样做：

from multiprocessing.dummy import Pool as ThreadPool

【讨论】：

您的导入完全倒退了。 multiprocessing.dummy.Pool 是基于线程的池，不是基于进程的池，multiprocessing.Pool 是基于进程的，不是基于线程的。

【解决方案2】：

您可以使用multiprocessing。

import multiprocessing

def string_fun(string):
    # do something with string...
    print(string)

my_strings = generate_random_strings()
num_of_threads = 7
pool = multiprocessing.Pool(num_of_threads)
pool.map(string_fun, my_strings)

【讨论】：

这段代码每个进程只传递一个字符串吗？如果我们要通过一大块它会更快吗？只是想解决这个问题。
@LoicDuros 根据文档，它正是这样做的——“这种方法将可迭代对象分割成许多块，作为单独的任务提交给进程池。” (docs.python.org/3.4/library/…)
好的，谢谢，如果生成器产生了大量的项目，是否需要“阅读”它才能执行 pool.map
我问的原因是因为当我只使用 4 个字符排列时，它实际上是从 string_fun 输出东西，但如果我切换到更高的数字，它看起来就像卡住了。
没关系，这是由于 generate_strings 造成的。

【解决方案3】：

假设您使用的是最新版本的 Python，您可能想阅读一些有关 asyncio 模块的内容。由于 GIL 锁，多线程不容易实现：“在 CPython 中，全局解释器锁或 GIL 是一个互斥锁，用于保护对 Python 对象的访问，防止多个线程同时执行 Python 字节码。这个锁是必要的主要是因为 CPython 的内存管理不是线程安全的。”

因此，您可以在 Multiprocessing 上进行交换，或者如上所述，查看 asycio 模块。
asyncio — 异步 I/O > https://docs.python.org/3/library/asyncio.html

我会尽快将此答案与一些代码集成。
希望对您有所帮助，
嘿嘿

【讨论】：

我已经准备好了例子，你可以看看吗？
是的，很好:)
谢谢，我会研究 asyncio

【解决方案4】：

正如@Hele 提到的，asyncio 是最好的，这里有一个例子

代码

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# python 3.7.2

from asyncio import ensure_future, gather, run
import random

alphabet = 'ABCDEFGH'
size = 1000


async def generate():
    tasks = list()
    result = None

    for el in range(1, size):
        task = ensure_future(generate_one())
        tasks.append(task)

        result = await gather(*tasks)

    return list(set(result))


async def generate_one():
    return ''.join(random.choice(alphabet) for i in range(8))


if __name__ == '__main__':

    my_strings = run(generate())

    print(my_strings)

输出

['CHABCGDD', 'ACBGAFEB', ...

当然，你需要改进generate_one，这个变种很慢。

可以看源码here。

【讨论】：

谢谢！我会测试和比较