這個問題在這里已經有了答案： FastAPI 以串行方式而不是并行方式運行 api 呼叫 2 個答案 4 小時前關閉。
我創建了一個等待webhook信號的服務器，當有信號時，它會創建一個新行程來運行loop()函式，而在運行loop()函式時，我希望它異步呼叫函式printmessage()，所以它會運行下一行回圈函式沒有等待printmessage()函式完成處理，但出現以下錯誤，我該如何解決？
#main.py
import time
from fastapi import Request, FastAPI
import multiprocessing as mp
import uvicorn
import asyncio
async def printmessage(fruit):
print(fruit)
time.sleep(5)
async def loop(fruit):
while True:
task = asyncio.create_task(printmessage(fruit))
time.sleep(1)
fruit="apple"
if __name__ == '__main__':
print("PROGRAM LAUNCH...")
print("WEBHOOK RECEIVE READY...")
app = FastAPI()
@app.post("/webhook")
async def webhook(request : Request):
print("WEBHOOK RECEIVED")
p = mp.Process(target=loop,args=[fruit])
p.start()
print('done')
return 'WEBHOOK RECEIVED'
預期的輸出應該apple每 1 秒列印一次。
錯誤：
RuntimeWarning: coroutine 'loop' was never awaited
self._target(*self._args, **self._kwargs)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
我嘗試了以下方法來避免錯誤，但根本沒有輸出：
#main.py
import time
from fastapi import Request, FastAPI
import multiprocessing as mp
import uvicorn
import asyncio
async def printmessage(fruit):
print(fruit)
time.sleep(5)
async def loop(fruit):
while True:
task = asyncio.create_task(printmessage(fruit))
time.sleep(1)
def preloop(fruit):
asyncio.run(loop(fruit))
fruit="apple"
if __name__ == '__main__':
print("PROGRAM LAUNCH...")
print("WEBHOOK RECEIVE READY...")
app = FastAPI()
@app.post("/webhook")
async def webhook(request : Request):
print("WEBHOOK RECEIVED")
p = mp.Process(target=preloop,args=[fruit])
p.start()
print('done')
return 'WEBHOOK RECEIVED'
uj5u.com熱心網友回復：
這是使用多處理在新行程中呼叫異步函式的方法。在此代碼中，對 /webhook 的每個請求都會創建一個新行程，該行程每 5 秒列印一次apple 。
from __future__ import annotations
import asyncio
from multiprocessing import Process
from fastapi import FastAPI
app = FastAPI()
process_pool: list[Process] = []
async def print_message(fruit):
print(fruit)
async def loop(fruit):
while True:
await print_message(fruit)
await asyncio.sleep(5)
def run_loop(fruit):
asyncio.run(loop(fruit))
@app.get("/webhook")
async def webhook():
print("WEBHOOK RECEIVED")
fruit = "apple"
process = Process(target=run_loop, args=(fruit,))
process_pool.append(process)
process.start()
print('done')
return 'WEBHOOK RECEIVED'
@app.on_event("shutdown")
async def shutdown_event():
for process in process_pool:
process.kill()
for process in process_pool:
while process.is_alive():
continue
process.close()
if __name__ == '__main__':
print("PROGRAM LAUNCH...")
print("WEBHOOK RECEIVE READY...")
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/453706.html
標籤：python-3.x 异步 蟒蛇异步 python-多处理 快速API
下一篇：結合兩個流和呼叫方法