python - 如何在保持顺序的同时在生成器上使用线程？

Question

我有一个简单的代码，它对我试图加速的生成器中的每个项目运行一个 GET 请求:

def stream(self, records):
    # type(records) = <type 'generator'>
    for record in records:
        # record = OrderedDict([('_time', '1518287568'), ('data', '5552267792')])
        output = rest_api_lookup(record[self.input_field])

        record.update(output)
        yield record

现在，它在单个线程上运行并且会一直运行，因为每个 REST 调用都会等待上一个 REST 调用完成。

在使用这个很好的答案(https://stackoverflow.com/a/28463266/1150923)之前，我已经在列表中使用了 Python 中的多线程，但我不确定如何在生成器而不是列表上重新使用相同的策略。

我从一位开发人员那里得到了一些建议，他建议我将生成器分解为 100 个元素的列表，然后关闭池，但我不知道如何从生成器创建这些列表。

我还需要保留原始顺序，因为我需要以正确的顺序yield record。

Answer 1

我假设您不想先将生成器 records 转换为列表。加快处理速度的一种方法是将记录按 block 传递到 ThreadPoolExecutor 中。执行器将同时处理您的 rest_api_lookup block 的所有项目。然后你只需要“unchunk”你的结果。这是一些运行示例代码(没有使用类，抱歉，但我希望它能说明原理):

from concurrent.futures import ThreadPoolExecutor
from time import sleep

pool = ThreadPoolExecutor(8) # 8 threads, adjust to taste and # of cores

def records():
    # simulates records generator
    for i in range(100):
        yield {'a': i}


def rest_api_lookup(a):
    # simulates REST call :)
    sleep(0.1)
    return {'b': -a}


def stream(records):
    def update_fun(record):
        output = rest_api_lookup(record['a'])
        record.update(output)
        return record
    chunk = []
    for record in records:
        # submit update_fun(record) into pool, keep resulting Future
        chunk.append(pool.submit(update_fun, record))
        if len(chunk) == 8:
            yield chunk
            chunk = []
    if chunk:
        yield chunk

def unchunk(chunk_gen):
    """Flattens a generator of Future chunks into a generator of Future results."""
    for chunk in chunk_gen:
        for f in chunk:
            yield f.result() # get result from Future

# Now iterate over all results in same order as generated by records()    
for result in unchunk(stream(records())):
    print(result)

喂!

更新:我在模拟的 REST 调用中添加了一个 sleep，以使其更加真实。这个分块版本在我的机器上在 1.5 秒内完成。顺序版本需要 10 秒(正如预期的那样，100 * 0.1s = 10s)。