在异步接口里面接收批量上传的文件夹后通过webdav3进行批量进行文件处理。其实涉及的问题就是:相对于在异步之中进行线程化的异步处理。
大致的代码如下所示:
@uploadrp.post('/upload/rp')
async def upload_rp_file(file: List[UploadFile] = File(...)):
........
upload_result = await new_upload_file(file_path=file_path, file_name=file_name, old_name=rp_dir_name)
.........
其中对应的new_upload_file也是异步函数,但是里面涉及到webdav3库的异步的调用,如下代码所示:
async def new_upload_file(file_path, file_name, old_name):
""" 上传文件夹 """
.........
upload_rp_file_result = client.upload_async('' + file_name, file_path,
callback=partial(completion_callback, old_file_dir=old_name, file_name=file_name))
.........
容纳后在调用upload_async的使用,需要进行任务处理完成的回调,也就是completion_callback函数,它的代码如下:
def completion_callback(old_file_dir, file_name):
pss
它的问题就是在completion_callback还有一个异步的函数需要调用的时候就遇到问题了:
(1) 我无法在upload_async上传文件完成后的回调函数使用async,即 async def completion_callback
(2) 我在completion_callback中获取当前事件循环,因为执行这个回调函数的时候,控制台打印当前没有事件循环
3) 我在completion_callback中new一个事件循环(new_event_loop)然后用create_task或run_until_complete执行update_rp_file_status函数的时候,都不成功:
- 要不然就报update_rp_file_status函数跟new的事件循环不是同一个...
- 要不就是这个update_rp_file_status压根不执行...
- 要不就是还没执行,new出来的事件循环,在update_rp_file_status还没执行的时候就被销毁了,很多问题...
问题探究
其实这个问题本质其实就是关于 client.upload_async中启用了新的线程去异步处理了,导致了开启了新的线程的问题,如下 client.upload_async的代码:
由此可见,问题肯本原因在于是因为我们是在新线程中调用了异步函数,而新线程没有自己的事件循环导致的。
回到事件循环的本身上,我们知道首先需要知道一点:
异步函数需要在事件循环中运行。但是由于我们启动服务之后,在主线程中,FastAPI应用程序已经创建了一个事件循环,并且通过uvicorn.run()方法来运行该应用程序,而这个的事件循环对象是当前主线程创建出来的, 而当我们在新线程中,也就是client.upload_async调用之后,开启的线程中并没有默认的事件循环可用。因此,当你尝试在新线程中调用异步函数时,就会抛出RuntimeError: There is no current event loop in thread异常,甚至你尝试打印获取当前运行的事件循环都无法获取到。如下代码所示:
from fastapi import FastAPI
import asyncio
app = FastAPI()
from threading import Thread # 创建线程的模块
def task(name):
print(asyncio.get_event_loop())
async def do_task():
# 异步任务的逻辑
await asyncio.sleep(1)
print("异步任务完成")
@app.get("/items")
async def read_item():
p = Thread(target=task, args=('线程1',))
p.start()
return {"data": "ok"}
在上面的等待吗中我们直接输出当前 print(asyncio.get_event_loop())也会遇到如下类似的错误提示:
问题解决
解决此类的问题,开始的时候,因为具体可能没了解带业务细节和逻辑,所以给的思路也有点偏差,后来自己尝试后,其实才发现原来只是一个比较简单的问题,此类的问题,我们可以在新线程中创建一个新的事件循环,并将任务添加到该事件循环中,以确保异步函数能够正确运行。这样,每个线程都有自己的事件循环,可以独立地运行异步任务。也就是如下的代码:
from fastapi import FastAPI
import asyncio
app = FastAPI()
from threading import Thread # 创建线程的模块
def task(name):
print(asyncio.get_event_loop())
# 尝试去执行异步任务
asyncio.run(do_task())
async def do_task():
# 异步任务的逻辑
await asyncio.sleep(1)
print("异步任务完成")
@app.get("/items")
async def read_item():
p = Thread(target=task, args=('线程1',))
p.start()
return {"data": "ok"}
在上面的代码中使用 asyncio.run开启新的事件循环去处理异步函数即可。但是之前那个朋友也说它尝试够使用在completion_callback中new一个事件循环(new_event_loop)然后用create_task或run_until_complete执行update_rp_file_status函数的时候,都不成功,或许可能得原因是它创建新的事件循环之后,没有进行重新设置,所以会导致无法设置成功,也就是我们也可以改为如下代码进行运行:
def task(name):
loop = asyncio.new_event_loop() # 创建新的事件循环
asyncio.set_event_loop(loop) # 设置新的事件循环为当前线程的事件循环
loop.run_until_complete(do_task()) # 在新的事件循环中运行异步任务
关于asyncio.run 和手动的设置创建事件循环
asyncio.run()是从Python 3.7版本引入的一个方便的函数,它用于运行异步函数。它自身会自动创建一个新的事件循环,并将异步函数添加到该事件循环中运行。在异步函数完成后,它会关闭事件循环并返回结果。而我们的
loop = asyncio.new_event_loop() # 创建新的事件循环
asyncio.set_event_loop(loop) # 设置新的事件循环为当前线程的事件循环
loop.run_until_complete(do_task()) # 在新的事件循环中运行异步
是一种手动创建、设置和运行事件循环的方法包括:
使用asyncio.new_event_loop()创建一个新的事件循环。使用asyncio.set_event_loop(loop)将新创建的事件循环设置为当前线程的事件循环。使用loop.run_until_complete(do_task())在新的事件循环中运行异步任务,直到任务完成。
他们之间的区别在于使用asyncio.run()时,它会自动处理事件循环的创建、设置和关闭,非常方便。而手动创建、设置和运行事件循环需要更多的代码来处理这些步骤,但也提供了更多的灵活性和控制权。