树莓派Python代码线程异常：线程无法正常关闭、捕获失效

问题描述 #

• 程序退出时出现线程未正常关闭的警告
• 按下Ctrl+C触发KeyboardInterrupt时，自定义的print内容无法输出，异常未生效

问题原因 #

1. KeyboardInterrupt仅发送给主线程：子线程中的try-except无法捕获该异常，导致中断信号无法被处理
2. 子线程内的join操作导致死锁：server和readBtn线程互相调用对方的join，会陷入无限等待
3. 主线程未处理退出逻辑：启动子线程后主线程直接结束，无法捕获中断信号并协调子线程退出

修正方案 #

核心修改点 #

• 让主线程负责捕获KeyboardInterrupt，统一管理子线程的退出和资源清理
• 子线程移除不必要的try-except块，仅通过event判断是否需要退出循环
• 给UDP的recvfrom添加超时，避免线程因阻塞无法响应退出信号
• 主线程中等待子线程全部结束后再执行GPIO清理

修正后的完整代码 #

import socket
import time
import RPi.GPIO as GPIO
import threading

event = threading.Event()
GPIO.setmode(GPIO.BCM)

# 调整LED状态存储结构，更清晰
leds_with_state = {
    0: {"pin": 21, "state": False}
}

# 初始化LED引脚
for led in leds_with_state.values():
    GPIO.setup(led["pin"], GPIO.OUT)
    GPIO.output(led["pin"], led["state"])

def server():
    buffer_size = 1024
    server_ip = '192.168.2.5'
    server_port = 9999
    rpi_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    rpi_server.bind((server_ip, server_port))
    # 设置超时，避免recvfrom一直阻塞
    rpi_server.settimeout(1)
    print("Server ready")

    while not event.is_set():
        try:
            cmd, address = rpi_server.recvfrom(buffer_size)
            cmd = cmd.decode('utf-8').strip()
            print(f"Received command: {cmd}")
            # 原代码的split操作无实际作用，可根据需求调整
            # cmd.split("<")[1]
            # cmd.split(">")[0]
            if cmd == "0":
                change_led(0)
        except socket.timeout:
            # 超时后继续循环，检查event状态
            continue
    # 关闭socket
    rpi_server.close()
    print("Server thread exited")

def read_btn():
    btn_pin = 20
    toggle_btn = 1
    GPIO.setup(btn_pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    while not event.is_set():
        read_val = GPIO.input(btn_pin)
        if read_val != toggle_btn and toggle_btn == 0:
            change_led(0)
        toggle_btn = read_val
        # 添加短延时，降低CPU占用
        time.sleep(0.05)
    print("Button thread exited")

def change_led(led_id):
    led = leds_with_state[led_id]
    led["state"] = not led["state"]
    GPIO.output(led["pin"], led["state"])
    print(f"LED {led_id} state changed to {led['state']}")

if __name__ == "__main__":
    server_thread = threading.Thread(target=server)
    read_btn_thread = threading.Thread(target=read_btn)
    # 设置为守护线程，防止主线程退出后子线程残留（可选，配合event更可靠）
    server_thread.daemon = True
    read_btn_thread.daemon = True

    server_thread.start()
    read_btn_thread.start()

    try:
        # 主线程阻塞等待中断
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nReceived exit signal, shutting down...")
        event.set()
        # 等待子线程结束
        server_thread.join()
        read_btn_thread.join()
        GPIO.cleanup()
        print("Cleanup done, program exited")

关键修改说明 #

1. 主线程捕获中断：将KeyboardInterrupt的处理移到主线程，确保中断信号被正确捕获
2. UDP超时设置：给recvfrom添加1秒超时，让server线程能定期检查event状态，及时响应退出指令
3. 移除子线程内的join和异常捕获：避免死锁，子线程仅通过event判断循环是否继续
4. 优化LED状态存储：改用字典结构，代码可读性更高
5. 添加延时降低CPU占用：read_btn线程中添加短延时，避免无意义的高频轮询
6. 守护线程设置：可选，确保主线程退出时子线程能自动终止（配合event双重保障）

内容的提问来源于stack exchange，提问作者Boyd Van Kronenburg