如何在p中运行多个阻塞IO协程

Question

我有一个用协同程序实现的长时间运行的管道。这通常是获取日志流，执行一些扩展（线程化）并将它们写入数据存储。在

下面是一个模拟管道的小示例：

import time
import random
from concurrent import futures

def coroutine(func):
    def start(*args, **kwargs):
        cr = func(*args, **kwargs)
        next(cr)
        return cr
    return start

@coroutine
def foo():
    pool = futures.ThreadPoolExecutor(max_workers=10)
    while True:
        i = (yield)
        fut = pool.submit(enrich, i)
        fut.add_done_callback(result_handler)
        time.sleep(random.random()*10)

def enrich(i):
    enriched = 'foo' + str(i)
    time.sleep(random.random())
    return enriched

def source(name, target):
    while True:
        time.sleep(random.random())
        i = random.randint(0,10)
        target.send(name + str(i))

下面调用的单个管道可以正常工作。在

^{pr2}$

现在，我想在后台线程中为不同的日志运行多个管道。一种尝试是再次使用ThreadPoolExecutor来驱动多个管道。在

def run():
    pool = futures.ThreadPoolExecutor(max_workers=10)
    tasks = [source('task one ', foo()),
             source('task two ', foo())]
    for task in tasks:
        fut = pool.submit(task)
        fut.add_done_callback(result_handler)

然而，在第一个任务之后，第二个任务将永远无法执行。在后台线程中运行如此长时间（可能永远）运行管道的正确方法是什么？在

Answer 1

因为source函数是永无止境的，所以不会创建tasks = [source('task one ', foo()), source('task two ', foo())]列表。这就是为什么第一个任务运行，管道阻塞。
解决方案是将source及其参数传递给pool.submit。在

tasks = [(source, 'task one', foo()), (source, 'task two', foo())]
for task in tasks:
    fut = pool.submit(*task)
    fut.add_done_callback(result_handler)