Python multiprocessing Event无法重复触发问题排查

两个核心问题导致事件只能触发一次：

• 事件消费逻辑没有循环：turn_left、turn_right两个子进程的目标函数里，wait()只被调用了一次。第一次事件触发、执行完动作、调用clear()之后，函数就直接执行完毕，对应子进程直接退出。后续生产进程check_light_dir即使再次调用set()，也没有存活的进程在等待该事件，自然不会触发动作，也不会有代码执行clear()重置标志位，所以你会看到事件标志一直卡在True状态。
• 代码遗漏导入：你使用了random.randint但没有加import random语句，运行时会抛出NameError，你贴的运行日志没有捕获到这个报错而已。

给两个事件消费函数加上无限循环，每次处理完动作后重新进入等待状态即可，修复后完整代码如下：

import time
import random
from multiprocessing import Event, Process

def check_light_dir(turn_left_e, turn_right_e):
    while True:
        time.sleep(2)
        var = random.randint(1, 10)
        print(f"var: {var}")
        if var < 5:
            print(f"turn_left_e before set(): {turn_left_e.is_set()}")
            turn_left_e.set()
        else:
            print(f"turn_right_e before set(): {turn_right_e.is_set()}")
            turn_right_e.set()


def turn_left(turn_left_e):
    while True:
        turn_left_e.wait()
        print(f"turn_left_e after set(): {turn_left_e.is_set()}")
        print("Turning servo left")
        turn_left_e.clear()
        print(f"turn_left_e after clear(): {turn_left_e.is_set()}")


def turn_right(turn_right_e):
    while True:
        turn_right_e.wait()
        print(f"turn_right_e after set(): {turn_right_e.is_set()}")
        print("Turning servo right")
        turn_right_e.clear()
        print(f"turn_right_e after clear(): {turn_right_e.is_set()}")


def test():
    turn_right_e = Event()
    turn_left_e = Event()

    check_light_dir_p = Process(target=check_light_dir, args=(turn_left_e, turn_right_e,))
    turn_right_p = Process(target=turn_right, args=(turn_right_e,))
    turn_left_p = Process(target=turn_left, args=(turn_left_e,))

    check_light_dir_p.start()
    turn_left_p.start()
    turn_right_p.start()

    check_light_dir_p.join()
    turn_right_p.join()
    turn_left_p.join()


if __name__ == "__main__":
    print("start")
    test()
    print("stop")

修复后运行就可以看到每次随机到对应方向时，伺服动作都会正常触发，事件标志也会在处理完成后被正常重置。

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