3、ROS2通信机制进阶：自定义消息与服务接口、QoS配置与最佳实践、同步与异步通信模式、多线程与回调组管理、通信延迟与带宽优化

好，咱们进入第三章。这一章，说白了就是让你真正「玩转」ROS2的通信。基础的消息收发谁都会，但到了实际项目中，你会发现事情远没那么简单。消息格式不够用怎么办？通信老是丢数据怎么办？程序跑起来卡顿怎么办？嗯，这一章就是来解决这些问题的。

3.1 自定义消息与服务接口

ROS2自带的那些标准消息类型，比如std_msgs/String、geometry_msgs/Pose，应付简单场景还行。但一旦你的机器人有特殊需求，比如要传一个「带时间戳的超声波传感器数组」，你就得自己动手了。

我个人习惯，把所有自定义接口文件放在一个单独的包里面，比如叫robot_custom_interfaces。这样管理起来清爽，别的包依赖也方便。

先看一个自定义消息的例子。假设我们要定义一个包含多个传感器数据的消息：

# UltrasonicArray.msg
std_msgs/Header header
float32[] distances
float32 temperature
uint8 sensor_count

然后在CMakeLists.txt里注册一下：

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/UltrasonicArray.msg"
  "srv/GetCalibration.srv"
)

服务接口也类似。比如我们要一个获取校准参数的服务：

# GetCalibration.srv
string sensor_id
---
float32 offset
float32 scale
bool success

小提示：接口文件里的字段类型，尽量用ROS2内置的原始类型，比如float32、int32。别自己搞复杂结构，否则后面序列化性能会吃亏。我在项目中见过有人用嵌套多层string数组，结果通信延迟直接翻倍。

3.2 QoS配置与最佳实践

QoS（Quality of Service）是ROS2里一个非常重要的概念。说白了，它就是告诉系统：「这个数据，你该怎么传、传丢了怎么办、要不要存历史记录。」

ROS2默认的QoS设置是RMW_QOS_PROFILE_DEFAULT，但实际项目中我很少直接用默认值。为什么？因为不同场景对可靠性的要求完全不同。

举个例子，激光雷达的数据，丢几帧没关系，但延迟必须低。这时候用SensorDataQoS：

from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy

sensor_qos = QoSProfile(
    depth=5,
    reliability=ReliabilityPolicy.BEST_EFFORT,
    durability=DurabilityPolicy.VOLATILE
)
self.publisher_ = self.create_publisher(LaserScan, 'scan', sensor_qos)

而控制指令呢？丢一条可能机器人就撞墙了。必须用可靠传输：

cmd_qos = QoSProfile(
    depth=10,
    reliability=ReliabilityPolicy.RELIABLE,
    durability=DurabilityPolicy.TRANSIENT_LOCAL
)

核心原则：传感器数据用BEST_EFFORT，控制指令用RELIABLE。这个原则我踩过坑——有一次用默认QoS传控制指令，结果WiFi波动时丢了几条，机器人直接原地打转。从那以后，我对QoS配置再也不敢马虎。

3.3 同步与异步通信模式

ROS2里，同步和异步通信的选择，直接影响你程序的响应速度和资源占用。

同步通信，就是客户端发请求后，一直等着服务端回复。适合那种「不拿到结果就不往下走」的场景，比如请求地图数据：

from rclpy.callback_groups import MutuallyExclusiveCallbackGroup
from rclpy.executors import SingleThreadedExecutor

client = self.create_client(GetMap, 'get_map')
while not client.wait_for_service(timeout_sec=1.0):
    self.get_logger().info('等待地图服务...')
future = client.call_async(request)
rclpy.spin_until_future_complete(self, future)
response = future.result()

异步通信则更灵活。你发个请求，然后该干嘛干嘛，等结果回来了再处理。适合那些不着急的任务，比如定期上传日志：

future = client.call_async(request)
future.add_done_callback(self.handle_response)

我个人习惯，能用异步就别用同步。同步调用会阻塞当前线程，一旦服务端响应慢，整个节点就卡住了。你想想看，导航节点因为等一个地图服务响应，连避障都停了，多危险。

3.4 多线程与回调组管理

ROS2默认是单线程执行器，所有回调按顺序排队执行。如果你的节点里既有耗时操作（比如图像处理），又有实时性要求高的操作（比如电机控制），单线程肯定不行。

这时候就需要多线程执行器和回调组了。

回调组有两种：

  • MutuallyExclusiveCallbackGroup：组内回调互斥，同一时间只能执行一个。适合保护共享资源。
  • ReentrantCallbackGroup：组内回调可以并发执行。适合无状态、可重入的操作。

看个实际例子。我有一个机器人底盘节点，既要处理速度指令（实时性高），又要定期发布里程计数据（可以慢一点）：

from rclpy.executors import MultiThreadedExecutor
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup, ReentrantCallbackGroup

# 高速回调组：处理速度指令
high_speed_group = MutuallyExclusiveCallbackGroup()
self.create_subscription(Twist, 'cmd_vel', self.cmd_callback, 10, callback_group=high_speed_group)

# 低速回调组：发布里程计
low_speed_group = ReentrantCallbackGroup()
self.create_timer(0.1, self.publish_odom, callback_group=low_speed_group)

# 使用多线程执行器
executor = MultiThreadedExecutor(num_threads=4)
executor.add_node(self)

注意：多线程环境下，共享数据一定要加锁。我曾经在里程计回调里直接修改了一个全局变量，结果速度指令回调也在读同一个变量，数据全乱了。后来老老实实加了threading.Lock才解决。

3.5 通信延迟与带宽优化

最后聊聊性能优化。ROS2的DDS底层虽然已经很高效了，但用不好照样卡成PPT。

延迟优化：

  • 减少消息大小：能传int32就别传float64，能用数组就别用嵌套结构。
  • 调整depth参数：队列深度太大，内存占用高；太小，容易丢消息。一般传感器设5-10，控制指令设10-20。
  • 使用零拷贝传输：ROS2支持unique_ptr传递消息，避免数据复制。适合大图像、点云数据。
auto msg = std::make_unique<sensor_msgs::msg::Image>();
// 填充数据...
publisher_->publish(std::move(msg));  // 零拷贝

带宽优化：

  • 降低发布频率：比如IMU数据从100Hz降到50Hz，肉眼几乎看不出区别，带宽省一半。
  • 使用压缩传输：ROS2支持CompressedImage等压缩消息类型。我做过测试，JPEG压缩后带宽能降到原来的十分之一。
  • 合理设置分区：不同功能的节点放在不同DDS分区，减少无关数据的网络传输。

实战经验：有一次做多机器人协同，三台机器人互相传点云数据，WiFi直接爆了。后来我把点云压缩成PointCloud2的binary格式，发布频率从30Hz降到10Hz，带宽从80Mbps降到了5Mbps。嗯，有时候「降频」比「优化代码」更管用。

好了，这一章的内容就到这里。通信机制是ROS2的骨架，把自定义接口、QoS、线程模型和性能优化都搞明白，你的机器人系统才能真正「跑起来、跑得稳、跑得快」。下一章我们聊聊更高级的话题——ROS2的组件化编程与生命周期管理。

← 上一章 📖 返回目录 下一章 →