先前有公眾號朋友問起一個問題，大概的問題是這樣：

在異步接口里面接收批量上傳的文件夾后通過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()時，它會自動處理事件循環的創建、設置和關閉，非常方便。而手動創建、設置和運行事件循環需要更多的代碼來處理這些步驟，但也提供了更多的靈活性和控制權。