如何在ROS Python类中使用回调函数数据？是否需多线程？

最简便的方案：类成员变量共享 #

ROS Python节点默认是单线程调度，所有回调函数和主逻辑都在同一个线程中执行，因此不需要额外多线程，直接通过类的实例变量共享数据即可。

核心思路：

• 在类的__init__中定义一个变量（比如self.latest_data）用于存储回调接收的数据
• 回调函数直接将消息数据赋值给这个类成员
• 主逻辑循环中读取该成员变量即可使用数据

代码示例：

import rospy
from std_msgs.msg import Float64  # 根据你的消息类型替换

class move_xarm:
    def __init__(self):
        # 初始化存储回调数据的变量，初始值设为None避免未接收数据时报错
        self.latest_joint_feedback = None
        # 初始化发布器
        self.cmd_pub = rospy.Publisher('/xarm/joint_command', Float64, queue_size=10)
        # 初始化订阅器，绑定回调函数
        self.feedback_sub = rospy.Subscriber('/xarm/joint_feedback', Float64, self.feedback_callback)

    def feedback_callback(self, msg):
        # 回调中直接更新类成员变量
        self.latest_joint_feedback = msg.data

    def run(self):
        # 主循环，按固定频率执行
        rate = rospy.Rate(10)  # 10Hz
        while not rospy.is_shutdown():
            # 检查是否已收到数据
            if self.latest_joint_feedback is not None:
                # 这里可以直接使用回调获取的数据，比如根据反馈调整指令
                adjusted_cmd = self.latest_joint_feedback + 0.05
                self.cmd_pub.publish(adjusted_cmd)
                rospy.loginfo(f"当前关节反馈值: {self.latest_joint_feedback}")
            # 让出CPU时间给回调函数执行，必须保留这一行
            rate.sleep()

if __name__ == '__main__':
    rospy.init_node('xarm_controller')
    xarm = move_xarm()
    xarm.run()

注意事项：

• 主循环中必须使用rate.sleep()或rospy.spin_once()，否则回调函数永远不会被执行（主循环会占满CPU）
• 回调函数要尽量简洁，不要在里面做耗时操作，否则会导致消息队列积压

何时需要多线程？ #

只有当你的主逻辑包含长时间阻塞操作（比如耗时的运动规划、硬件阻塞调用等）时，才需要使用多线程。因为单线程下阻塞操作会卡住整个节点，导致回调无法及时处理新消息。

核心思路：

• 使用Python的threading模块将主逻辑放到子线程
• 主线程专门执行rospy.spin()处理回调
• 用线程锁（threading.Lock）保证多线程下数据读写的安全性

代码示例：

import rospy
import threading
from std_msgs.msg import Float64

class move_xarm:
    def __init__(self):
        self.latest_joint_feedback = None
        # 初始化线程锁，防止多线程读写冲突
        self.data_lock = threading.Lock()
        self.cmd_pub = rospy.Publisher('/xarm/joint_command', Float64, queue_size=10)
        self.feedback_sub = rospy.Subscriber('/xarm/joint_feedback', Float64, self.feedback_callback)

    def feedback_callback(self, msg):
        # 写数据时加锁
        with self.data_lock:
            self.latest_joint_feedback = msg.data

    def main_logic(self):
        rate = rospy.Rate(10)
        while not rospy.is_shutdown():
            # 读数据时加锁
            with self.data_lock:
                current_feedback = self.latest_joint_feedback
            
            if current_feedback is not None:
                # 这里可以执行耗时操作，不会阻塞回调
                rospy.loginfo(f"处理反馈值: {current_feedback}")
                # 模拟耗时操作（比如运动规划）
                # time.sleep(1)
                self.cmd_pub.publish(current_feedback + 0.05)
            rate.sleep()

    def run(self):
        # 启动主逻辑子线程
        logic_thread = threading.Thread(target=self.main_logic)
        logic_thread.start()
        # 主线程持续处理回调
        rospy.spin()
        # 等待子线程结束
        logic_thread.join()

if __name__ == '__main__':
    rospy.init_node('xarm_controller')
    xarm = move_xarm()
    xarm.run()

关键避坑点 #

• 永远不要在回调函数中执行耗时操作，回调的职责只是接收并存储数据
• 多线程场景下必须使用线程锁保护共享变量，避免数据竞争
• 初始化时给共享变量设置合理的初始值（比如None），防止未接收数据时访问报错

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