我有一個主線程，它始終需要可用于讀取鍵盤按鍵輸入。每次按鍵時，我必須運行 4 個功能，例如顯示、保存到文件以及發出 HTTP 請求等。按鍵的速度比執行 4 個功能的速度更快。我的邏輯如下，但無法看到如何保持 key_press() 始終運行，因為我必須等待函數在某個時刻完成。async def main(): async with aiohttp.ClientSession() as client: while True: pressed_key = key_press() if pressed_key != "": f1 = asyncio.create_task(do_something1()) f2 = asyncio.create_task(do_something2()) f3 = asyncio.create_task(do_something3()) f4 = asyncio.create_task(send_to_API()) await f1,f2,f3,send_to_APIasyncio.run(main()) 我是否應該嘗試一些額外的邏輯，在遞歸調用 asyncio.tasks 后進行更多循環？或者我應該考慮多線程，讓一個線程活動運行 key_press() 并在其他地方進行異步調用？有什么想法或暗示朝著好的方向發展嗎？
1 回答
哆啦的時光機
TA貢獻1779條經驗 獲得超6個贊
如果您嘗試借助input（key_press示例中的函數）模擬USB數據流，則必須使用multithreading模塊，因為它input是一個阻塞函數，它將停止asyncio線程中的循環工作。要asyncio與input函數結合，您必須使用另一個線程，請查找以下示例：
import asyncio
import threading
async def do_work(i):
"""take input key and do some work with the input"""
await asyncio.sleep(5)
print(i)
async def main_thread_loop_work(_key):
"""simulate multiple tasks caused by key input"""
t1 = asyncio.create_task(do_work(_key))
t2 = asyncio.create_task(do_work(_key))
t3 = asyncio.create_task(do_work(_key))
await t1
await t2
await t3
def thead_worker(_key):
"""target function for another thread"""
asyncio.run(main_thread_loop_work(_key))
if __name__ == '__main__':
while True:
some_key = input("Please provide any key: ")
th = threading.Thread(target=thead_worker, args=(some_key, ))
th.start()
添加回答
舉報
0/150
提交
取消