如何在异步代码中处理 CPU 密集型任务答案

【问题标题】：How to process a CPU-bound task in async code如何在异步代码中处理 CPU 密集型任务
【发布时间】：2023-02-07 14:43:36
【问题描述】：

我正在做一些需要异步方法的繁重处理。我的一个方法返回一个字典列表，在将其添加到另一个可等待对象之前需要经过大量处理。 IE。

def cpu_bound_task_here(record):
    ```some complicated preprocessing of record```
    return record

在好心人给出下面的答案后，我的代码现在就卡住了。

async def fun():
print("Socket open")
record_count = 0
symbol = obj.symbol.replace("-", "").replace("/", "")
loop = asyncio.get_running_loop()
await obj.send()

while True:
    try:
        records = await obj.receive()
        if not records:
            continue

        record_count += len(records)

所以上面的函数所做的，是它的异步流值，并在无限期地推送到 redis 之前做一些繁重的处理。我进行了必要的更改，现在我被卡住了。

【问题讨论】：

标签： python-asyncio

【解决方案1】：

正如该输出告诉您的那样，run_in_executor 返回 Future。您需要等待它才能得到结果。

record = await loop.run_in_executor(
    None, something_cpu_bound_task_here, record
)

请注意，something_cpu_bound_task_here 的任何参数都需要传递给run_in_executor。

此外，正如您所提到的，这是一项受 CPU 限制的任务，您需要确保使用的是 concurrent.futures.ProcessPoolExecutor。除非您在某处调用了loop.set_default_executor，否则默认是ThreadPoolExecutor 的一个实例。

with ProcessPoolExecutor() as executor:
    for record in records:
        record = await loop.run_in_executor(
            executor, something_cpu_bound_task_here, record
        )

最后，您的 while 循环有效地同步运行。您需要等待未来，然后等待 obj.add，然后再继续处理 records 中的下一项。您可能想要稍微重组您的代码并使用 gather 之类的东西来允许一些并发。

async def process_record(record, obj, loop, executor):
    record = await loop.run_in_executor(
        executor, something_cpu_bound_task_here, record
    )
    await obj.add(record)

async def fun():
    loop = asyncio.get_running_loop()
    records = await receive()
    with ProcessPoolExecutor() as executor:
        await asyncio.gather(
            *[process_record(record, obj, loop, executor) for record in records]
        )

我不确定如何处理 obj，因为您的示例中未定义它，但我相信您可以解决这个问题。

【讨论】：

哎，我的obj其实是一个redis流。它来自 aioredis 库，我试图将记录添加到此流。但在添加记录之前，我正在对这些记录进行一些复杂的预处理。
我更改了问题以更好地反映我实际尝试做的事情
嘿，我试过你的方法，但现在它导致我的代码中断。在这里粘贴我的代码，好奇你是否可以帮忙？
我的代码现在由于某种原因卡住了
它卡在哪里？知道这一点会让你更容易告诉你哪里出了问题。

【解决方案2】：

查看库Pypeln，它非常适合在进程、线程和异步池之间流式传输任务：

import pypeln as pl
data = get_iterable()
data = pl.task.map(f1, data, workers=100) # asyncio
data = pl.thread.flat_map(f2, data, workers=10)
data = filter(f3, data)
data = pl.process.map(f4, data, workers=5, maxsize=200)

【讨论】：