Python多线程实现双LDR读取故障排查求助

我用PiMyLife教程里的代码，通过电容读取单个光敏电阻（LDR）数据时运行正常，但添加第二个LDR和电容到引脚31，复制rc_time函数及调用逻辑后，第二个函数始终无法执行。尝试用Python多线程实现，第二个线程还是跑不起来，求排查原因。

初始扩展代码 #

#!/usr/local/bin/python

import RPi.GPIO as GPIO
import time
GPIO.setmode(GPIO.BOARD)

# 定义电路连接的引脚
pin_ldr_one = 29
pin_ldr_two = 31

def rc_time_one(pin_ldr_one):
    count_one = 0
  
    # 引脚设为输出并拉低
    GPIO.setup(pin_ldr_one, GPIO.OUT)
    GPIO.output(pin_ldr_one, GPIO.LOW)
    time.sleep(0.1)

    # 引脚切换为输入
    GPIO.setup(pin_ldr_one, GPIO.IN)
  
    # 计数直到引脚变高
    while (GPIO.input(pin_ldr_one) == GPIO.LOW):
        count_one += 1

    return count_one


def rc_time_two(pin_ldr_two):
    count_two = 0
  
    # 引脚设为输出并拉低
    GPIO.setup(pin_ldr_two, GPIO.OUT)
    GPIO.output(pin_ldr_two, GPIO.LOW)
    time.sleep(0.1)

    # 引脚切换为输入
    GPIO.setup(pin_ldr_two, GPIO.IN)
  
    # 计数直到引脚变高
    while (GPIO.input(pin_ldr_two) == GPIO.LOW):
        count_two += 1

    return count_two
    
# 捕获中断信号，正确清理资源
try:
    # 主循环
    while True:
        print("LDR 1: ", rc_time_one(pin_ldr_one))
        print("LDR 2: ", rc_time_two(pin_ldr_two))
except KeyboardInterrupt:
    pass
finally:
    GPIO.cleanup()

多线程实现代码 #

#!/usr/local/bin/python

import RPi.GPIO as GPIO
import time
from time import sleep, perf_counter
from threading import Thread

GPIO.setmode(GPIO.BOARD)

# 定义电路连接的引脚
pin_ldr_one = 29
pin_ldr_two = 31

def rc_time_one(pin_ldr_one = 29):
    count_one = 0
  
    # 引脚设为输出并拉低
    GPIO.setup(pin_ldr_one, GPIO.OUT)
    GPIO.output(pin_ldr_one, GPIO.LOW)
    time.sleep(0.1)

    # 引脚切换为输入
    GPIO.setup(pin_ldr_one, GPIO.IN)
  
    # 计数直到引脚变高
    while (GPIO.input(pin_ldr_one) == GPIO.LOW):
        count_one += 1
    print("LDR 1: ", count_one)
    return count_one


def rc_time_two(pin_ldr_two = 31):
    count_two = 0
  
    # 引脚设为输出并拉低
    GPIO.setup(pin_ldr_two, GPIO.OUT)
    GPIO.output(pin_ldr_two, GPIO.LOW)
    time.sleep(0.1)

    # 引脚切换为输入
    GPIO.setup(pin_ldr_two, GPIO.IN)
  
    # 计数直到引脚变高
    while (GPIO.input(pin_ldr_two) == GPIO.LOW):
        count_two += 1
    print("LDR 2: ", count_two)
    return count_two
    
# 捕获中断信号，正确清理资源
try:
    # 主循环
    while True:
        # 创建两个线程
        t1 = Thread(target=rc_time_one)
        t2 = Thread(target=rc_time_two)

        # 启动线程
        t1.start()
        t2.start()
        t1.join()
        t2.join()
except KeyboardInterrupt:
    pass
finally:
    GPIO.cleanup()

核心原因 #

1. GPIO线程安全问题：RPi.GPIO库本身不支持线程安全，多线程同时调用GPIO.setup、GPIO.input等操作会触发资源冲突，导致第二个线程的GPIO操作被阻塞或失败。
2. 无限循环阻塞（初始代码）：若第二个LDR电路存在硬件问题（如电容未正确放电、接线错误），rc_time_two中的while循环会无限执行，函数永远无法返回，后续代码也无法运行。
3. 多线程伪并行：你写的多线程代码用t1.join()等待第一个线程完成后才处理第二个线程，本质还是串行执行，且线程安全问题未解决。

修复步骤 #

步骤1：硬件排查 #

• 检查第二个LDR接线：确认引脚31正确连接到LDR与电容的串联电路（电容接地、LDR接3.3V，中间节点接引脚31）。
• 单独测试第二个LDR：编写仅读取引脚31的代码，确认硬件本身无故障。

步骤2：给初始代码添加超时机制 #

避免因硬件故障导致的无限循环：

def rc_time_two(pin_ldr_two):
    count_two = 0
    timeout = time.time() + 2  # 设置2秒超时
  
    GPIO.setup(pin_ldr_two, GPIO.OUT)
    GPIO.output(pin_ldr_two, GPIO.LOW)
    time.sleep(0.1)

    GPIO.setup(pin_ldr_two, GPIO.IN)
  
    while (GPIO.input(pin_ldr_two) == GPIO.LOW):
        count_two += 1
        if time.time() > timeout:
            print("LDR 2 读取超时")
            return -1  # 返回错误标记

    return count_two

步骤3：线程安全的多线程实现 #

用线程锁保证同一时间只有一个线程操作GPIO：

#!/usr/local/bin/python

import RPi.GPIO as GPIO
import time
from threading import Thread, Lock

GPIO.setmode(GPIO.BOARD)
pin_ldr_one = 29
pin_ldr_two = 31

# 创建GPIO操作锁
gpio_lock = Lock()

def rc_time(pin, ldr_name):
    count = 0
    timeout = time.time() + 2
  
    with gpio_lock:  # 加锁，避免GPIO操作冲突
        GPIO.setup(pin, GPIO.OUT)
        GPIO.output(pin, GPIO.LOW)
        time.sleep(0.1)
        GPIO.setup(pin, GPIO.IN)
  
    # 读取引脚状态无需加锁
    while (GPIO.input(pin) == GPIO.LOW):
        count += 1
        if time.time() > timeout:
            print(f"{ldr_name} 读取超时")
            return -1
    print(f"{ldr_name}: {count}")
    return count
    
try:
    while True:
        t1 = Thread(target=rc_time, args=(pin_ldr_one, "LDR 1"))
        t2 = Thread(target=rc_time, args=(pin_ldr_two, "LDR 2"))

        t1.start()
        t2.start()
        t1.join()
        t2.join()
        time.sleep(0.5)  # 避免循环过于频繁
except KeyboardInterrupt:
    pass
finally:
    GPIO.cleanup()

步骤4：替代方案——单线程轮询+超时 #

如果多线程仍有问题，单线程轮询是更可靠的选择，彻底规避线程安全问题：

#!/usr/local/bin/python

import RPi.GPIO as GPIO
import time

GPIO.setmode(GPIO.BOARD)
pin_ldr_one = 29
pin_ldr_two = 31

def read_ldr(pin, timeout=2):
    count = 0
    start_time = time.time()
  
    GPIO.setup(pin, GPIO.OUT)
    GPIO.output(pin, GPIO.LOW)
    time.sleep(0.1)
    GPIO.setup(pin, GPIO.IN)
  
    while GPIO.input(pin) == GPIO.LOW:
        count += 1
        if time.time() - start_time > timeout:
            return -1
    return count
    
try:
    while True:
        ldr1_val = read_ldr(pin_ldr_one)
        ldr2_val = read_ldr(pin_ldr_two)
        print(f"LDR 1: {ldr1_val}, LDR 2: {ldr2_val}")
        time.sleep(0.5)
except KeyboardInterrupt:
    pass
finally:
    GPIO.cleanup()

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