【发布时间】：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" 端点。
【问题讨论】：
标签： python asynchronous fastapi uvicorn starlette