树莓派4中gpiozero与asyncio协同异常：仅首个协程启动

问题根源 #

你的代码存在两个核心问题，导致只有第一个协程能运行：

1. rotary_detection中的asyncio.sleep(0.01)未添加await关键字，这个异步函数不会触发协程挂起，导致该协程持续占用事件循环，无法让其他协程执行。
2. button_detection的while True循环没有任何异步挂起操作，一旦启动就会无限阻塞事件循环，完全剥夺其他协程的运行机会。

解决方案1：修复轮询式协程 #

给两个协程中的异步操作添加await，让事件循环能够调度多个协程：

import asyncio
from gpiozero import RotaryEncoder, Button

button = Button(7)
encoder = RotaryEncoder(a=10, b=9, wrap=False, max_steps=0)

async def rotary_detection():
    print("rotary_detection started")
    last_rotary_value = 0
    try:
        while True:
            current_rotary_value = encoder.steps
            if last_rotary_value != current_rotary_value:
                rotDiff = last_rotary_value - current_rotary_value
                print("turned CW" if rotDiff > 0 else "turned CCW")
                last_rotary_value = current_rotary_value
            await asyncio.sleep(0.01)  # 添加await，交出事件循环控制权
    except KeyboardInterrupt:
        print("Program terminated")
    except Exception as e:
        print(e)

async def button_detection():
    print("button_detection started")
    try:
        while True:
            if button.is_pressed:
                print("Button pressed")
                await asyncio.sleep(0.2)  # 短延迟避免重复触发
            await asyncio.sleep(0.01)  # 添加await，交出事件循环控制权
    except KeyboardInterrupt:
        print("Program terminated")
    except Exception as e:
        print(e)

async def main():
    await asyncio.gather(
        rotary_detection(),
        button_detection()
    )

if __name__ == "__main__":
    asyncio.run(main())

解决方案2：使用GPIOZero事件回调（更高效） #

GPIOZero本身支持事件回调机制，无需轮询，后台线程触发回调，与Asyncio兼容，资源占用更低：

import asyncio
from gpiozero import RotaryEncoder, Button

button = Button(7)
encoder = RotaryEncoder(a=10, b=9, wrap=False, max_steps=0)

last_rotary_value = 0

def on_rotary_turned():
    global last_rotary_value
    current_rotary_value = encoder.steps
    rotDiff = last_rotary_value - current_rotary_value
    print("turned CW" if rotDiff > 0 else "turned CCW")
    last_rotary_value = current_rotary_value

def on_button_pressed():
    print("Button pressed")

async def main():
    global last_rotary_value
    last_rotary_value = encoder.steps
    
    # 绑定事件回调
    encoder.when_rotated = on_rotary_turned
    button.when_pressed = on_button_pressed
    
    # 保持事件循环持续运行
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        print("Program terminated")

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者Doemelti