Python异步进程协调：双传感器数据支撑机器人控制中心

问题背景 #

基于Python开发移动机器人，采用RPLIDAR实现避障、UWB实现定位。单独测试时：

• get_data_from_LIDAR()处理耗时2-20ms
• uwb_data_visualization()处理耗时1300-2100ms
  二者处理速度差异极大，属于典型异步场景。曾尝试互相嵌套调用，导致RPLIDAR抛出"Incorrect starting descriptor bytes and wrong body size"错误，程序卡顿——核心原因是UWB的慢处理阻塞了LIDAR的实时数据采集流程。

核心需求 #

让两个传感器数据处理函数独立运行，各自更新controlCenter()所需的共享变量：

• LIDAR提供的避障数据（angle340_359、angle0_20等）
• UWB提供的定位及任务状态数据（x1、y1、is_delivery_complete等）

解决方案：多线程+线程安全共享变量 #

利用Python的threading模块实现两个传感器函数的独立运行，同时通过线程锁（threading.Lock）保证共享变量的读写安全，避免多线程冲突。

步骤1：初始化线程锁与共享变量 #

在robotController的__init__方法中添加锁和共享变量的初始化逻辑：

import threading
import time
from math import floor
import sys
import RPi.GPIO as GPIO
# 导入RPLIDAR、UWB、turtle等依赖库

class robotController:
    def __init__(self):
        # 保留原传感器连接初始化代码
        # self.lidar = ... 
        # self.uwb_device = ...

        # 线程锁，保护共享变量读写操作
        self.data_lock = threading.Lock()

        # LIDAR共享避障数据
        self.angle340_359 = []
        self.angle0_20 = []
        self.angle270_339 = []
        self.angle21_90 = []
        self.cycle = 0
        self.con_scan = True

        # UWB共享定位及状态数据
        self.x1 = 0.0
        self.y1 = 0.0
        self.is_delivery_complete = False
        self.lastExecTime = 0
        self.elapsedTime = 50  # UWB数据更新间隔
        self.meter2pixel = 100  # 自定义米转像素系数
        self.distance_a1_a2 = 2.0  # 锚点A1782与A1783的间距

        # 保留原turtle可视化相关初始化代码
        # self.t_a1 = ...
        # self.t_a2 = ...
        # self.t_a3 = ...

步骤2：修改传感器函数，添加线程安全操作 #

对两个传感器函数进行改造，在读写共享变量时通过锁保证线程安全：

改造get_data_from_LIDAR

def get_data_from_LIDAR(self):
        while self.con_scan:
            try:
                # 移除频繁打印LIDAR信息的代码，降低输出开销
                for scan in self.lidar.iter_scans(max_buf_meas=500):
                    scan_data = [0]*360
                    for (_, angle, distance) in scan:
                        intAngle = floor(angle)
                        index = min(359, intAngle)
                        scan_data[index] = distance

                    # 加锁更新共享避障数据
                    with self.data_lock:
                        self.angle340_359 = scan_data[340:360]
                        self.angle0_20 = scan_data[0:21]
                        self.angle270_339 = scan_data[270:340]
                        self.angle21_90 = scan_data[21:91]
                        self.cycle += 1

            except KeyboardInterrupt:
                self.con_scan = False
                print("\nLIDAR Stopped!")
                self.lidar.stop()
                self.lidar.disconnect()
                GPIO.cleanup()
                sys.exit()
            except Exception as e:
                print(f"LIDAR Error: {e}")
                # 出错后短暂休眠，避免无限循环报错
                time.sleep(0.1)

改造uwb_data_visualization

def uwb_data_visualization(self):
        while True:
            curTime = int(time.time()*1000)
            if (curTime - self.lastExecTime) > self.elapsedTime:
                node_count = 0
                uwb_list = self.read_data()  # 避免用list作为变量名

                a1_range = 0.0
                a2_range = 0.0
                for one in uwb_list:
                    if one["A"] == "1782":
                        self.clean(self.t_a1)
                        a1_range = self.uwb_range_offset(float(one["R"]))
                        self.draw_uwb_anchor(-250, 150, "A1782(0,0)", a1_range, self.t_a1)
                        node_count += 1
                    if one["A"] == "1783":
                        self.clean(self.t_a2)
                        a2_range = self.uwb_range_offset(float(one["R"]))
                        self.draw_uwb_anchor(-250 + self.meter2pixel * self.distance_a1_a2, 150, 
                                            f"A1783({self.distance_a1_a2})", a2_range, self.t_a2)
                        node_count += 1

                if node_count == 2:
                    x1, y1 = self.tag_pos(a2_range, a1_range, self.distance_a1_a2)
                    print(f"x1: {x1}m  y1: {y1}m")
                    self.clean(self.t_a3)
                    self.draw_uwb_tag(x1, y1, "TAGi", self.t_a3)

                    x2 = 1.2
                    y2 = 1.4
                    distance_between_centers = ((x2 - x1)**2 + (y2 - y1)**2)**0.5
                    distance_threshold = 0.75

                    is_complete = False
                    if distance_between_centers < distance_threshold:
                        self.draw_line(x1,y1,x2,y2,color="yellow")
                        is_complete = True
                    else:
                        self.draw_line(x1,y1,x2,y2,color="black")

                    # 加锁更新共享定位数据
                    with self.data_lock:
                        self.x1 = x1
                        self.y1 = y1
                        self.is_delivery_complete = is_complete

                self.lastExecTime = curTime
            # 短暂休眠，降低CPU占用率
            time.sleep(0.01)
        # 将turtle.mainloop()移至主线程，避免阻塞当前线程

步骤3：改造controlCenter，线程安全读取共享变量 #

def controlCenter(self):
        while True:
            # 加锁读取共享数据，避免读取过程中数据被修改
            with self.data_lock:
                is_complete = self.is_delivery_complete
                front_left_dist = max(self.angle340_359) if self.angle340_359 else 0
                front_right_dist = max(self.angle0_20) if self.angle0_20 else 0

            # 电机控制逻辑
            if is_complete:
                print("stop")
            else:
                print("forward")

            if front_left_dist >= 650 or front_right_dist >= 650:
                print("rotateCW")
            else:
                print("keep moving")

            # 控制循环频率，避免空转消耗CPU
            time.sleep(0.02)

步骤4：主线程启动多线程 #

if __name__ == "__main__":
    robot = robotController()

    # 创建传感器线程，设置daemon=True保证主线程退出时子线程自动终止
    lidar_thread = threading.Thread(target=robot.get_data_from_LIDAR, daemon=True)
    uwb_thread = threading.Thread(target=robot.uwb_data_visualization, daemon=True)

    # 启动线程
    lidar_thread.start()
    uwb_thread.start()

    # 若需要turtle可视化，将mainloop放在主线程
    # turtle.mainloop()

    # 启动控制中心
    robot.controlCenter()

关键注意事项 #

• 线程安全：所有共享变量的读写必须通过with self.data_lock:包裹，防止多线程同时操作导致的数据混乱。
• 守护线程：设置daemon=True，确保主线程退出时子线程自动结束，避免资源泄漏。
• CPU占用优化：在循环中添加短暂休眠，避免空转消耗过多系统资源。
• 错误处理：LIDAR函数中添加出错后的休眠逻辑，防止异常发生时频繁打印错误信息。
• 可视化线程隔离：turtle.mainloop()会阻塞线程，建议放在主线程或单独线程，避免影响传感器数据处理流程。

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