如何用asyncIO实现mido音序器的启停切换控制?
问题根源与解决方案
核心问题分析
- 同步Sleep阻塞事件循环:代码中使用的
time.sleep(0.001)是同步阻塞函数,会冻结整个异步事件循环,导致音序器播放暂停,同时阻塞后续MIDI消息的处理。 - 协程阻塞导致消息无法处理:
toggleSequencer中直接await runSequencer(),而runSequencer是一个无限循环协程,会让toggleSequencer一直挂起。第一次按下按钮后,readInport中的await asyncio.gather(*queue)会一直等待toggleSequencer完成,后续按钮消息根本没机会被处理。
具体修改方案
1. 替换同步Sleep为异步Sleep
在__main__.py中,把同步的time.sleep换成异步的asyncio.sleep,避免阻塞事件循环:
async def readInport() -> None: for msg in inport.iter_pending(): if msg.type != 'sysex': type, ch, note, vel = functions.unpack_message(msg) print(type, ch, note, vel) queue = [] if type == "control_change" and note == 19 and vel == 127: await asyncio.sleep(0.001) # 替换为异步sleep queue = [seq.toggleSequencer() for seq in sequencers] else: print("other") await asyncio.gather(*queue)
2. 重构音序器启停逻辑
修改sequencer.py,让runSequencer作为后台任务运行,避免阻塞toggleSequencer:
async def runSequencer(self) -> None: while self.isRunning: # 音序器核心代码(发送MIDI消息等) await asyncio.sleep(0.1) # 必须加异步sleep,给事件循环处理其他任务的时间 async def toggleSequencer(self) -> None: print(self.voice, "TOGGLE") self.isRunning = not self.isRunning if self.isRunning: # 创建后台任务运行音序器,无需await,避免阻塞当前协程 asyncio.create_task(self.runSequencer()) # 暂停时仅需切换状态,runSequencer的循环会自动退出 await asyncio.sleep(0.001)
3. 优化主循环
给main函数添加短暂的异步sleep,确保事件循环有时间处理所有任务:
async def main(): while True: await readInport() await asyncio.sleep(0.001) if __name__ == "__main__": asyncio.run(main())
关键注意事项
runSequencer的循环内必须添加await asyncio.sleep(...),哪怕是极短的延时,否则该协程会独占事件循环,导致MIDI消息读取、其他任务无法执行。self.isRunning在异步环境下无需额外线程锁,因为所有异步任务都在同一线程的事件循环中串行执行,不存在并发修改问题。
内容的提问来源于stack exchange,提问作者Adir
相关产品推荐
相关产品推荐

