关于FastAPI中在新线程里调用协成函数问题

2024年 1月 4日 172.8k 0

先前有公众号朋友问起一个问题,大概的问题是这样:

在异步接口里面接收批量上传的文件夹后通过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()时,它会自动处理事件循环的创建、设置和关闭,非常方便。而手动创建、设置和运行事件循环需要更多的代码来处理这些步骤,但也提供了更多的灵活性和控制权。

相关文章

JavaScript2024新功能:Object.groupBy、正则表达式v标志
PHP trim 函数对多字节字符的使用和限制
新函数 json_validate() 、randomizer 类扩展…20 个PHP 8.3 新特性全面解析
使用HTMX为WordPress增效:如何在不使用复杂框架的情况下增强平台功能
为React 19做准备:WordPress 6.6用户指南
如何删除WordPress中的所有评论

发布评论