如何异步处理Python keyboard库全局热键的回调阻塞问题?
问题原因
你之前用asyncio改造失效的核心原因是 asyncio.run() 是阻塞调用,每次热键触发都会启动一个独立的新事件循环,且必须等当前协程完全执行完才会返回,本质还是同步执行回调,自然会阻塞后续热键的触发。
单线程asyncio实现方案
核心思路
全程只运行一个常驻的asyncio事件循环,热键触发时直接将回调协程提交到事件循环的任务队列,提交完成后立即返回,不需要等待协程执行结束,同时把keyboard的热键监听逻辑和asyncio事件循环整合到同一个线程调度。
可运行测试代码
import asyncio import keyboard async def on_callback(key): print(f'pressed {key}') await asyncio.sleep(5) # 模拟异步长耗时任务 print(f'end for {key}') async def main(): # 注册热键 keyboard.add_hotkey("f4", lambda: asyncio.create_task(on_callback("f4"))) keyboard.add_hotkey("f8", lambda: asyncio.create_task(on_callback("f8"))) # 异步监听退出信号,不阻塞事件循环 while not keyboard.is_pressed('esc'): await asyncio.sleep(0.01) if __name__ == "__main__": asyncio.run(main())
运行后依次按下f4、f8,即可得到期望的输出:
pressed f4 pressed f8 end for f4 end for f8
业务代码适配修改点
你当前业务中的Google API调用是同步IO,会阻塞事件循环,需要做对应调整:
- 优先将
googleapiclient替换为异步实现的Google API客户端,把AppsScript.run改造为异步方法,所有IO操作使用await调用异步接口,实现纯单线程无阻塞。 - 如果暂时不想改造同步HTTP逻辑,可以用
asyncio.to_thread把同步调用包装为异步执行(底层由asyncio线程池调度,开销远低于手动创建新线程),修改示例:
async def _on_callback(self, function, keys): response = await asyncio.to_thread(self._script.run, function) # 后续处理response的逻辑
内容的提问来源于stack exchange,提问作者Viewed
相关产品推荐
相关产品推荐

