ROS2多线程编程实战：Python与C++高效并发处理

1. 项目概述

作为一名长期从事机器人开发的工程师，我经常遇到需要同时处理多个任务的场景。比如在ROS2中，我们可能需要一边接收传感器数据，一边控制执行器运动，同时还要处理用户输入。这时候，多线程编程就成了必备技能。

今天我要分享的是ROS2环境下Python和C++的多线程实战经验。不同于教科书式的理论讲解，我会结合具体机器人开发场景，带你从零开始构建可落地的多线程解决方案。无论你是刚接触ROS2的新手，还是有一定经验的开发者，这篇文章都能帮你避开我当年踩过的那些坑。

2. 多线程编程基础

2.1 为什么ROS2需要多线程

在机器人系统中，很多任务都是并行发生的。以移动机器人为例：

• 激光雷达以10Hz频率发布点云数据
• 电机控制器需要以100Hz频率发送控制指令
• 导航算法需要以5Hz频率更新路径规划

如果全部放在单线程中顺序执行，要么高频任务会被低频任务阻塞，要么低频任务得不到及时处理。这就是我们需要多线程的根本原因。

2.2 Python vs C++多线程特性对比

在ROS2中，我们主要使用Python和C++两种语言。它们的多线程实现有显著差异：

特性	Python (threading)	C++ (std::thread)
真并行	受GIL限制	是
内存共享	容易(需注意GIL)	需要显式同步
创建开销	较小	较大
适用场景	I/O密集型	计算密集型

提示：Python由于GIL(全局解释器锁)的存在，多线程在CPU密集型任务中性能提升有限。但在ROS2中，大部分时间都在等待I/O(如网络消息)，所以Python多线程仍然很实用。

3. Python多线程实战

3.1 基础线程创建

让我们从一个简单的Python示例开始：

python复制import threading
import time

def sensor_task():
    while True:
        print("采集传感器数据...")
        time.sleep(0.1)  # 模拟耗时操作

def control_task():
    while True:
        print("执行控制指令...")
        time.sleep(0.2)

if __name__ == '__main__':
    t1 = threading.Thread(target=sensor_task)
    t2 = threading.Thread(target=control_task)
    
    t1.start()
    t2.start()
    
    t1.join()
    t2.join()

这个例子创建了两个线程，分别模拟传感器数据采集和控制指令执行。注意几个关键点：

1. threading.Thread创建线程对象
2. start()方法启动线程
3. join()等待线程结束

3.2 ROS2中的线程安全实践

在ROS2中直接使用多线程时，需要特别注意rclpy的线程安全性。下面是一个安全的ROS2 Python多线程示例：

python复制import rclpy
from rclpy.node import Node
import threading

class MultiThreadedNode(Node):
    def __init__(self):
        super().__init__('multi_threaded_node')
        self.sensor_thread = threading.Thread(target=self.sensor_loop)
        self.control_thread = threading.Thread(target=self.control_loop)
        
    def sensor_loop(self):
        while rclpy.ok():
            self.get_logger().info("处理传感器数据")
            time.sleep(0.1)
            
    def control_loop(self):
        while rclpy.ok():
            self.get_logger().info("执行控制逻辑")
            time.sleep(0.2)
            
    def run(self):
        self.sensor_thread.start()
        self.control_thread.start()
        
        try:
            while rclpy.ok():
                time.sleep(0.5)
        except KeyboardInterrupt:
            pass
            
        self.sensor_thread.join()
        self.control_thread.join()

def main():
    rclpy.init()
    node = MultiThreadedNode()
    node.run()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

关键注意事项：

1. 每个线程中都要检查rclpy.ok()
2. 使用节点的get_logger()而不是直接print
3. 主线程负责处理信号和异常

4. C++多线程实战

4.1 std::thread基础用法

C++11引入了std::thread，让多线程编程变得简单。看一个基本示例：

cpp复制#include <iostream>
#include <thread>
#include <chrono>

void sensorTask() {
    while (true) {
        std::cout << "采集传感器数据..." << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void controlTask() {
    while (true) {
        std::cout << "执行控制指令..." << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
}

int main() {
    std::thread t1(sensorTask);
    std::thread t2(controlTask);
    
    t1.join();
    t2.join();
    
    return 0;
}

与Python版本的主要区别：

1. 使用std::thread直接构造线程
2. 时间单位更精确(chrono)
3. 不需要像Python那样显式调用start()

4.2 ROS2 C++多线程最佳实践

在ROS2 C++节点中使用多线程时，我们需要考虑更复杂的资源管理。下面是一个完整示例：

cpp复制#include "rclcpp/rclcpp.hpp"
#include <thread>
#include <atomic>

class MultiThreadedNode : public rclcpp::Node {
public:
    MultiThreadedNode() : Node("multi_threaded_