在后台任务中使用多个工作人员 - Fast-API

Posted 2023-03-08

【中文标题】在后台任务中使用多个工作人员 - Fast-API

【英文标题】：Using several workers in a background task - Fast-API

【发布时间】：2020-12-06 23:40:35

【问题描述】：

我正在尝试处理用户上传的文件。但是，我希望用户在上传完成后得到响应并终止连接但继续处理文件。因此，我使用的是 BackgroundTasks.add_tasks，我的代码如下所示：

class Line(BaseModel):
    line: str

@app.post("/foo")
async def foo(line: Line):
""" Processing line generate results"""

    ...

    result = ... # processing line.line
    print(results)
    return results

@app.post("/upload")
async def upload(background_tasks: BackgroundTasks, csv: UploadFile = File(...)):

    background_tasks.add_task(process, csv)
    return response.text("CSV has been uploaded successfully")


async def process(csv):
    """ Processing CSV and generate data"""

    tasks = [foo(line) for line in csv]
    result = await asyncio.gather(*tasks)

不幸的是，上面的代码只能一个接一个地执行。此外，我必须等到所有结果都处理完毕，然后 foo 中的打印语句才起作用，即假设我在 csv 中有 n 行，在处理完所有 n 之后是当我看到打印语句时对所有人。我的程序在 20 个工作人员上运行，但是当这个进程运行时，它只使用了大约 1% 的 CPU（foo 不是计算任务，它更像是一个 IO/网络绑定任务）。这让我认为后台进程仅在 1 个工作人员上运行。我确实尝试了 ProcessPoolExecutor 如下：

loop = asyncio.get_event_loop()
lines = [line_0, line_1, ..., line_n] # Extracted all lines from CSV
with ProcessPoolExecutor() as executor:
    results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines]
    results = loop.run_until_complete(*results)

但是，我收到以下错误：

processpoolexecutor 不能腌制本地对象

我确实通过改变方法克服了这个错误 来自：

results = [loop.run_in_executor(executor, lambda: foo(line)) for line in lines]

到：

results = [asyncio.ensure_future(foo(line=Line(line)) for line in lines]

但是，我得到了这个错误：

文件“uvloop/loop.pyx”，第 2658 行，在 uvloop.loop.Loop.run_in_executor AttributeError：“循环”对象没有属性“提交”

总结：要处理一行，我可以点击 "/foo" 端点。现在，我想处理 200 行的 csv。所以首先我接受来自用户的文件并返回成功消息并终止该连接。然后将 csv 添加到后台任务，该任务应将每一行映射到 "/foo" 端点并为我提供每一行的结果。但是，到目前为止，我尝试过的所有方法似乎都只使用一个线程，并且逐行处理每一行。我想要一种可以同时处理多行的方法，就像我们可以使用 Apache JMeter 等工具一样同时多次访问 "/foo" 端点。

【参考方案1】：

您可以在不使用端点的情况下进行并行处理。 下面是一个基于您的代码的简化示例（不使用foo 端点）：

import asyncio
import sys
import uvicorn
from fastapi import FastAPI, BackgroundTasks, UploadFile, File
from loguru import logger


logger.remove()
logger.add(sys.stdout, colorize=True, format="<green>time:HH:mm:ss</green> | level | <level>message</level>")

app = FastAPI()


async def async_io_bound(line: str):
    await asyncio.sleep(3)  # Pretend this is IO operations
    return f"Line 'line' processed"


async def process(csv):
    """ Processing CSV and generate data"""
    tasks = [async_io_bound(line) for line in csv]
    logger.info("start processing")
    result = await asyncio.gather(*tasks)
    for i in result:
        logger.info(i)


@app.post("/upload-to-process")
async def upload(background_tasks: BackgroundTasks, csv: UploadFile = File(...)):
    background_tasks.add_task(process, csv.file)
    return "result": "CSV has been uploaded successfully"

if __name__ == "__main__":
    uvicorn.run("app3:app", host="localhost", port=8001)

输出示例（所有行并行处理）：

INFO:     ::1:52358 - "POST /upload-to-process HTTP/1.1" 200 OK
13:21:31 | INFO | start processing
13:21:34 | INFO | Line 'b'one, two\n'' processed
13:21:34 | INFO | Line 'b'0, 1\n'' processed
13:21:34 | INFO | Line 'b'1, 1\n'' processed
13:21:34 | INFO | Line 'b'2, 1\n'' processed
13:21:34 | INFO | Line 'b'3, 1\n'' processed
13:21:34 | INFO | Line 'b'4, 1\n'' processed
13:21:34 | INFO | Line 'b'5, 1\n'' processed
13:21:34 | INFO | Line 'b'6, 1\n'' processed
13:21:34 | INFO | Line 'b'7, 1\n'' processed
13:21:34 | INFO | Line 'b'8, 1\n'' processed
13:21:34 | INFO | Line 'b'9, 1\n'' processed

【讨论】：

我确实尝试过实施您的建议。但是，这些进程仍在一个接一个地运行。因此，我不得不求助于 arq 来让进程并行运行。我使用 arq 而不是 celery，因为我有异步功能。截至本文发表时，Celery 不支持异步功能，它应该在 2020 年 12 月发布的最新版本 v5 中获得支持。

最好看看你的线处理的实现对你有帮助