Python中threading.Timer递归场景下如何通知调用方任务完成

首先，咱们得戳破核心问题：threading.Timer是异步执行的，递归调用turn()的时候，leftTurn()根本没法直接拿到它的返回值——毕竟Timer的代码跑在另一个线程，主线程的leftTurn()早就继续往下走了。所以得换个思路，用同步标记或者回调函数来让leftTurn()感知转向完成的时机。

下面给你两种实用的落地方案：

方案一：用threading.Event做完成标记 #

这是最直观的同步方式，用一个事件对象标记转向状态，leftTurn()只需要等待这个事件被触发就行，完美解决异步通知的问题。

修改后的代码示例：

import threading
from mpu6050 import mpu6050  # 假设你使用的是这个MPU6050库

class navigation():
    def __init__(self):
        self.mpu = mpu6050(0x68)
        self.turn_completed = threading.Event()  # 初始化完成事件标记
        self.total_turned_angle = 0  # 累计转向角度的变量

    def getOffset(self):
        gyro_data = self.mpu.get_gyro_data()
        offset = abs(gyro_data['z'])
        return offset

    def turn(self, target_angle, interval=0.01):
        # 计算本次Timer周期内的角度增量（陀螺仪积分：角速度×时间间隔）
        gyro_z = self.mpu.get_gyro_data()['z']
        angle_increment = gyro_z * interval
        self.total_turned_angle += abs(angle_increment)  # 累计左转角度（取绝对值）

        # 检查是否达到目标转向角度
        if self.total_turned_angle >= target_angle:
            self.turn_completed.set()  # 触发完成事件，通知leftTurn()
            return
        
        # 未完成则继续定时调用turn，用循环式Timer替代递归，避免栈溢出
        threading.Timer(interval, self.turn, args=(target_angle, interval)).start()

    def leftTurn(self, target_angle):
        # 每次转向前重置状态
        self.total_turned_angle = 0
        self.turn_completed.clear()
        
        # 启动转向流程
        self.turn(target_angle)
        
        # 同步等待转向完成
        self.turn_completed.wait()
        print("左转已完成！")
        return True  # 现在可以正常返回完成标记了

方案优势： #

• 逻辑直白，leftTurn()可以同步等待完成，不需要处理复杂的异步逻辑
• 用循环式Timer替代递归，彻底避免了长时间转向导致的递归栈溢出问题

方案二：用回调函数异步通知完成 #

如果你的场景需要异步执行（比如leftTurn()不想被阻塞，还要同时处理其他任务），可以给turn()传入一个回调函数，当转向完成时自动触发回调。

示例代码：

import threading
from mpu6050 import mpu6050

class navigation():
    def __init__(self):
        self.mpu = mpu6050(0x68)
        self.total_turned_angle = 0

    def getOffset(self):
        gyro_data = self.mpu.get_gyro_data()
        offset = abs(gyro_data['z'])
        return offset

    def turn(self, target_angle, complete_callback, interval=0.01):
        gyro_z = self.mpu.get_gyro_data()['z']
        angle_increment = gyro_z * interval
        self.total_turned_angle += abs(angle_increment)

        if self.total_turned_angle >= target_angle:
            complete_callback()  # 触发回调，通知转向完成
            return
        
        threading.Timer(interval, self.turn, args=(target_angle, complete_callback, interval)).start()

    def leftTurn(self, target_angle):
        self.total_turned_angle = 0
        
        # 定义转向完成后的回调逻辑
        def on_turn_finished():
            print("左转完成！")
            # 这里可以写完成后的后续操作，比如启动下一个导航任务
        
        # 启动转向，传入回调函数
        self.turn(target_angle, on_turn_finished)
        
        # 不需要同步等待的话，这里可以直接返回，回调会在后台触发

方案优势： #

• 适配异步场景，不会阻塞leftTurn()所在的主线程
• 灵活性高，回调函数里可以自定义任意完成后的业务逻辑

额外提醒： #

1. 陀螺仪积分误差：实际使用中，陀螺仪的漂移会让积分误差越来越大，建议结合加速度计/磁力计做互补滤波，或者定期用getOffset()校准零点偏移。
2. Timer精度问题：threading.Timer的定时精度不算特别高，如果需要更精准的控制，可以考虑用time.sleep()结合线程循环，或者异步框架asyncio的定时器（如果是异步项目）。

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