我正在尝试处理用户上传的文件。但是，我希望用户在上传完成后得到响应并终止连接但继续处理文件。因此，我使用的是 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: 'Loop' object has no attribute 'submit'
总结：要处理一行，我可以点击“/foo”端点。现在，我想处理 200 行的 csv。所以首先我接受来自用户的文件并返回成功消息并终止该连接。然后将 csv 添加到后台任务，该任务应将每一行映射到“/foo”端点并为我提供每一行的结果。但是，到目前为止我尝试过的所有方法似乎都只使用一个线程，并且正在逐行处理每一行。我想要一种可以同时处理多行的方法，就像我们可以使用 Apache JMeter 之类的工具一样同时多次点击“/foo”端点。