3. ROS2基础通信实战：话题、服务、动作的编程实现

大家好，我是你们的自动驾驶感知工程师讲师。今天咱们来聊聊ROS2里最核心的三个通信方式——话题、服务、动作。说实话，这三个东西搞明白了，ROS2开发你就入门了一大半。

我记得刚接触ROS2那会儿，总觉得通信机制有点绕。后来在项目里踩了不少坑，才慢慢摸清门道。今天我就把这些经验掰开揉碎了讲给你听。

3.1 话题（Topic）——最常用的异步通信

话题是ROS2里最基础的通信方式。说白了，就是发布者往一个“频道”里发数据，订阅者从这个“频道”里收数据。发布者和订阅者互不认识，完全解耦。

我在项目中遇到过一个问题：摄像头采集的图像数据量很大，如果用服务通信，每帧图像都要等响应，帧率根本跑不起来。用话题就完美解决了——发布者只管发，订阅者只管收，谁也不等谁。

3.1.1 发布者编程示例

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.pub = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(1.0, self.timer_callback)
        self.count = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello ROS2: {self.count}'
        self.pub.publish(msg)
        self.get_logger().info(f'发布: {msg.data}')
        self.count += 1

def main(args=None):
    rclpy.init(args=args)
    node = Talker()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

3.1.2 订阅者编程示例

class Listener(Node):
    def __init__(self):
        super().__init__('listener')
        self.sub = self.create_subscription(
            String, 'chatter', self.callback, 10)

    def callback(self, msg):
        self.get_logger().info(f'收到: {msg.data}')

你想想看，订阅者只需要知道话题名字和消息类型，就能收到数据。这种松耦合的设计，在自动驾驶系统里特别实用。比如激光雷达节点发布点云数据，感知模块、定位模块、规划模块都能同时订阅，各取所需。

3.2 服务（Service）——同步请求-响应

服务跟话题不一样。它是同步的，客户端发一个请求，服务端处理完再返回响应。说白了，就是一问一答的模式。

我曾经在调试一个相机标定程序时，需要临时查询相机参数。用话题的话，得一直监听，太浪费。用服务就刚刚好——需要的时候发个请求，拿到结果就完事。

3.2.1 服务端编程示例

from example_interfaces.srv import AddTwoInts

class ServiceServer(Node):
    def __init__(self):
        super().__init__('service_server')
        self.srv = self.create_service(
            AddTwoInts, 'add_two_ints', self.add_callback)

    def add_callback(self, request, response):
        response.sum = request.a + request.b
        self.get_logger().info(
            f'收到: {request.a} + {request.b} = {response.sum}')
        return response

3.2.2 客户端编程示例

class ServiceClient(Node):
    def __init__(self):
        super().__init__('service_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待服务...')
        self.req = AddTwoInts.Request()

    def send_request(self, a, b):
        self.req.a = a
        self.req.b = b
        future = self.cli.call_async(self.req)
        rclpy.spin_until_future_complete(self, future)
        return future.result()

3.3 动作（Action）——带反馈的长时间任务

动作是ROS2里最复杂的通信方式。它结合了话题和服务的优点——既有服务的请求-响应模式，又有话题的实时反馈能力。

嗯，这里要注意。动作特别适合自动驾驶里的规划控制任务。比如车辆变道，你发一个“变道”的动作请求，它一边执行一边给你反馈进度，执行完了告诉你结果。如果发现障碍物，还能中途取消。

3.3.1 动作定义文件

# action/Drive.action
int32 target_speed
int32 target_lane
---
bool success
string message
---
float32 progress

3.3.2 动作服务端编程示例

from rclpy.action import ActionServer
from your_package.action import Drive

class DriveActionServer(Node):
    def __init__(self):
        super().__init__('drive_action_server')
        self.action_server = ActionServer(
            self, Drive, 'drive', self.execute_callback)

    async def execute_callback(self, goal_handle):
        self.get_logger().info(f'收到目标: 速度={goal_handle.request.target_speed}')
        
        feedback_msg = Drive.Feedback()
        feedback_msg.progress = 0.0
        
        for i in range(10):
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                return Drive.Result(success=False, message='任务取消')
            
            feedback_msg.progress = (i + 1) * 10.0
            goal_handle.publish_feedback(feedback_msg)
            await asyncio.sleep(1.0)
        
        goal_handle.succeed()
        return Drive.Result(success=True, message='任务完成')

3.4 三种通信方式对比

特性	话题	服务	动作
通信模式	异步	同步	异步
是否支持反馈	否	否	是
是否可取消	否	否	是
适用场景	传感器数据、状态发布	参数查询、简单计算	导航、机械臂控制
实时性	高	中	中

3.5 知识体系结构图

3.6 实战经验总结

我个人习惯在项目开始前，先画一个通信拓扑图。把每个节点用什么通信方式标清楚，这样后面写代码不会乱。

我曾经在一个项目里，把本该用话题的传感器数据硬生生用成了服务。结果每帧图像都要等响应，帧率从30fps掉到了2fps。后来改成话题，问题立刻解决。你想想看，选对通信方式有多重要。

好了，三种通信方式就讲到这里。代码示例我都跑过，可以直接用。建议你动手敲一遍，遇到问题再回来翻翻。实践出真知嘛。