我有一個(gè)主線程，它始終需要可用于讀取鍵盤按鍵輸入。每次按鍵時(shí)，我必須運(yùn)行 4 個(gè)功能，例如顯示、保存到文件以及發(fā)出 HTTP 請(qǐng)求等。按鍵的速度比執(zhí)行 4 個(gè)功能的速度更快。我的邏輯如下，但無法看到如何保持 key_press() 始終運(yùn)行，因?yàn)槲冶仨毜却瘮?shù)在某個(gè)時(shí)刻完成。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()) 我是否應(yīng)該嘗試一些額外的邏輯，在遞歸調(diào)用 asyncio.tasks 后進(jìn)行更多循環(huán)？或者我應(yīng)該考慮多線程，讓一個(gè)線程活動(dòng)運(yùn)行 key_press() 并在其他地方進(jìn)行異步調(diào)用？有什么想法或暗示朝著好的方向發(fā)展嗎？
1 回答
哆啦的時(shí)光機(jī)
TA貢獻(xiàn)1779條經(jīng)驗(yàn) 獲得超6個(gè)贊
如果您嘗試借助input（key_press示例中的函數(shù)）模擬USB數(shù)據(jù)流，則必須使用multithreading模塊，因?yàn)樗黫nput是一個(gè)阻塞函數(shù)，它將停止asyncio線程中的循環(huán)工作。要asyncio與input函數(shù)結(jié)合，您必須使用另一個(gè)線程，請(qǐng)查找以下示例：
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()
添加回答
舉報(bào)
0/150
提交
取消