asyncio.run()阻塞子进程，舵机控制程序servoLoop未运行问题咨询

问题原因 #

你的代码存在三个核心问题：

1. 变量作用域错误：main函数中修改的servo0ang是局部变量，未使用global声明，导致全局的servo0ang从未被更新。
2. 多进程内存隔离：multiprocessing.Process创建的子进程会复制父进程初始内存空间，但后续两者内存完全独立。子进程中的servo0ang始终保持初始值90，父进程的修改无法传递到子进程。
3. 无CPU延迟的循环：servoLoop中的无限循环会占用100%CPU，可能导致进程调度异常，甚至硬件响应卡顿。

解决方案 #

方案一：使用线程替代进程（推荐） #

线程共享同一内存空间，无需额外通信机制，代码更简洁高效：

import asyncio
import sys
import threading
import time

# 请确保正确导入Servo类
# from your_servo_library import Servo

async def connect_stdin_stdout():
    loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, w_protocol = await loop.connect_write_pipe(asyncio.streams.FlowControlMixin, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop)
    return reader, writer

servo0ang = 90

async def main():
    global servo0ang
    reader, writer = await connect_stdin_stdout()
    while True:
        res = await reader.read(100)
        if not res:
            break
        # 处理输入异常，限制舵机角度范围
        try:
            angle = int(res.strip())
            servo0ang = max(0, min(180, angle))
        except ValueError:
            print("输入无效，请输入0-180之间的数字")

def servoLoop():
    pwm = Servo()
    while True:
        pwm.setServoPwm('0', servo0ang)
        time.sleep(0.01)  # 添加延迟，降低CPU占用

if __name__ == "__main__":
    # 守护线程：主程序退出时自动终止
    t = threading.Thread(target=servoLoop, daemon=True)
    t.start()
    asyncio.run(main())

方案二：保留多进程，使用共享内存 #

若必须使用进程，需通过multiprocessing.Value实现父子进程的变量共享：

import asyncio
import sys
import time
from multiprocessing import Process, Value

# 请确保正确导入Servo类
# from your_servo_library import Servo

async def connect_stdin_stdout():
    loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, w_protocol = await loop.connect_write_pipe(asyncio.streams.FlowControlMixin, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop)
    return reader, writer

async def main(servo0ang):
    reader, writer = await connect_stdin_stdout()
    while True:
        res = await reader.read(100)
        if not res:
            break
        try:
            angle = int(res.strip())
            servo0ang.value = max(0, min(180, angle))
        except ValueError:
            print("输入无效，请输入0-180之间的数字")

def servoLoop(servo0ang):
    pwm = Servo()
    while True:
        pwm.setServoPwm('0', servo0ang.value)
        time.sleep(0.01)

if __name__ == "__main__":
    # 创建共享整数变量，初始值90
    servo0ang = Value('i', 90)
    p = Process(target=servoLoop, args=(servo0ang,))
    p.start()
    asyncio.run(main(servo0ang))
    p.join()

额外注意事项 #

• 输入校验：必须捕获非数字输入的异常，避免程序崩溃。
• 角度限制：根据舵机实际参数限制角度范围，防止硬件损坏。
• 延迟优化：循环中添加短暂延迟，避免CPU资源被无限循环耗尽。

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