ROS2 提供了四种核心通信机制，分别适用于不同的应用场景：

• 话题（Topics）：异步数据流，最常用
• 服务（Services）：同步请求 - 响应
• 动作（Actions）：带反馈的长时间任务
• 参数（Parameters）：节点配置共享

一、底层基础：DDS 通信架构

所有 ROS2 通信都建立在 DDS 之上，理解 DDS 的核心概念是掌握 ROS2 通信的关键。

DDS 核心概念：

• 域（Domain）：DDS 的隔离单元，通过域 ID（Domain ID） 区分。只有相同域 ID 的节点才能通信，ROS2 默认域 ID 是 0。
• 参与者（Participant）：每个 ROS2 节点对应一个 DDS 参与者，是通信的入口。
• 主题（Topic）：数据的标识符，相同主题的发布者和订阅者才能匹配。
• 数据写入器（DataWriter）：发布者的底层实现，负责发送数据。
• 数据读取器（DataReader）：订阅者的底层实现，负责接收数据。

DDS 带来的核心优势：

1. 真正的分布式通信：没有 ROS1 那样的 Master 节点，节点之间直接 P2P 通信；
2. 丰富的 QoS 策略：可以精确控制通信的可靠性、延迟、优先级等；
3. 实时性保障：支持硬实时通信，满足工业机器人和自动驾驶的需求；
4. 跨平台跨语言：支持 Linux、Windows、macOS，以及 C++、Python、Java 等多种语言；

二、核心通信方式详解

2.1 话题（Topics）：发布 - 订阅模式

最常用的通信方式，适用于高频、连续、单向的数据流传输。

工作原理：

• 多个发布者（Publisher）可以向同一个话题发布消息；
• 多个订阅者（Subscriber）可以订阅同一个话题接收消息；
• 发布者和订阅者完全解耦，不知道对方的存在；
• 通信是异步的：发布者发送消息后立即返回，不等待订阅者处理；

特点：

• 通信模式：一对多、多对多；
• 方向性：单向（发布者→订阅者）；
• 同步性：异步；
• 数据类型：.msg 文件定义的结构化消息；

适用场景：

• 传感器数据传输（摄像头、激光雷达、IMU）；
• 机器人状态发布（关节角度、位置、速度）；
• 控制指令广播；
• 日志输出；

示例（Python）：

# 发布者
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class MinimalPublisher(Node):
    def __init__(self):
        super().__init__('minimal_publisher')
        self.publisher_ = self.create_publisher(String, 'topic', 10)
        timer_period = 0.5  # 0.5秒发布一次
        self.timer = self.create_timer(timer_period, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello ROS2: {self.i}'
        self.publisher_.publish(msg)
        self.get_logger().info(f'Publishing: {msg.data}')
        self.i += 1

# 订阅者
class MinimalSubscriber(Node):
    def __init__(self):
        super().__init__('minimal_subscriber')
        self.subscription = self.create_subscription(
            String,
            'topic',
            self.listener_callback,
            10)
        self.subscription  # 防止未使用变量警告

    def listener_callback(self, msg):
        self.get_logger().info(f'I heard: {msg.data}')

2.2 服务（Services）：请求 - 响应模式

适用于一次性、需要明确结果的双向通信。

工作原理：

• 客户端（Client）向服务端发送一个请求（Request）；
• 服务端（Server）处理请求后返回一个响应（Response）；
• 一个服务端可以被多个客户端调用，但同一时间只能处理一个请求；
• 通信默认是同步的：客户端发送请求后会阻塞等待响应；

特点：

• 通信模式：一对一；
• 方向性：双向（请求→响应）；
• 同步性：默认同步，支持异步调用；
• 数据类型：.srv 文件定义，分为 Request 和 Response 两部分；

适用场景

• 触发式操作（启动 / 停止电机、保存地图）；
• 参数查询和设置；
• 一次性计算（逆运动学求解、路径规划请求）；
• 节点生命周期管理；

示例（Python）：

# 服务端
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class MinimalService(Node):
    def __init__(self):
        super().__init__('minimal_service')
        self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_two_ints_callback)

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

# 客户端（同步调用）
class MinimalClient(Node):
    def __init__(self):
        super().__init__('minimal_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('Service not available, waiting again...')
        self.req = AddTwoInts.Request()

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

2.3 动作（Actions）：目标 - 反馈 - 结果模式

ROS2 中最强大也最容易被忽视的通信方式，专门为长时间运行的任务设计。

工作原理：

动作实际上是话题 + 服务的组合，包含三个独立的通信通道：

1. 目标通道（服务）：客户端向服务端发送一个目标；
2. 反馈通道（话题）：服务端在执行过程中持续向客户端发送进度反馈；
3. 结果通道（服务）：任务完成后，服务端向客户端返回最终结果；
4. 取消通道（服务）：客户端可以随时取消正在执行的任务；

特点：

• 通信模式：一对一；
• 方向性：双向；
• 同步性：异步；
• 数据类型：.action 文件定义，分为 Goal、Feedback 和 Result 三部分；
• 核心优势：可抢占、可取消、有进度反馈；

适用场景：

• 机械臂运动控制（移动到指定位置）；
• 机器人导航（从 A 点移动到 B 点）；
• 抓取任务；
• 任何需要知道执行进度或可以中断的任务；

# 动作服务器
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from example_interfaces.action import Fibonacci

class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__('fibonacci_action_server')
        self._action_server = ActionServer(
            self,
            Fibonacci,
            'fibonacci',
            self.execute_callback)

    def execute_callback(self, goal_handle):
        self.get_logger().info('Executing goal...')
        feedback_msg = Fibonacci.Feedback()
        feedback_msg.partial_sequence = [0, 1]

        for i in range(1, goal_handle.request.order):
            feedback_msg.partial_sequence.append(
                feedback_msg.partial_sequence[i] + feedback_msg.partial_sequence[i-1])
            self.get_logger().info(f'Feedback: {feedback_msg.partial_sequence}')
            goal_handle.publish_feedback(feedback_msg)
            # 模拟耗时操作
            time.sleep(1)

        goal_handle.succeed()
        result = Fibonacci.Result()
        result.sequence = feedback_msg.partial_sequence
        return result

# 动作客户端
class FibonacciActionClient(Node):
    def __init__(self):
        super().__init__('fibonacci_action_client')
        self._action_client = ActionClient(self, Fibonacci, 'fibonacci')

    def send_goal(self, order):
        goal_msg = Fibonacci.Goal()
        goal_msg.order = order
        self._action_client.wait_for_server()
        self._send_goal_future = self._action_client.send_goal_async(
            goal_msg,
            feedback_callback=self.feedback_callback)
        self._send_goal_future.add_done_callback(self.goal_response_callback)

    def feedback_callback(self, feedback_msg):
        feedback = feedback_msg.feedback
        self.get_logger().info(f'Received feedback: {feedback.partial_sequence}')

    def goal_response_callback(self, future):
        goal_handle = future.result()
        if not goal_handle.accepted:
            self.get_logger().info('Goal rejected')
            return
        self.get_logger().info('Goal accepted')
        self._get_result_future = goal_handle.get_result_async()
        self._get_result_future.add_done_callback(self.get_result_callback)

    def get_result_callback(self, future):
        result = future.result().result
        self.get_logger().info(f'Result: {result.sequence}')

2.4 参数（Parameters）：节点配置共享

参数本质上是基于服务实现的键值对存储，用于节点之间共享配置信息。

特点：

• 每个节点都有自己的参数服务器；
• 参数可以在节点启动时设置，也可以运行时动态修改；
• 支持多种数据类型：整数、浮点数、布尔值、字符串、数组；
• 可以设置参数的描述、范围和默认值；

适用场景：

• 节点配置参数（PID 参数、传感器校准参数）；
• 运行时动态调整参数；
• 全局配置共享；

三、三种核心通信方式对比

特性	话题（Topics）	服务（Services）	动作（Actions）
通信模式	发布 - 订阅	请求 - 响应	目标 - 反馈 - 结果
连接方式	一对多 / 多对多	一对一	一对一
同步性	异步	默认同步	异步
方向性	单向	双向	双向
反馈机制	无	只有最终结果	持续进度反馈
可取消性	不适用	不能取消	可以取消
适用场景	高频连续数据流	一次性请求响应	长时间运行任务
实时性	高	中	中

四、关键特性：QoS（服务质量）策略

QoS 是 ROS2 通信最核心的优势之一，它允许你精确控制通信的行为，以满足不同场景的需求。

常用 QoS 策略

1. 可靠性（Reliability）

   • BEST_EFFORT（最佳努力）：尽力发送，不保证消息到达。适用于传感器数据，实时性比可靠性更重要。
   • RELIABLE（可靠）：保证消息一定到达，丢失会重传。适用于控制指令和重要数据。

2. 历史记录（History）

   • KEEP_LAST：只保留最近 N 条消息（N 由 Depth 指定）。最常用。
   • KEEP_ALL：保留所有消息，直到被处理。可能导致内存溢出。

3. 深度（Depth）

   • 当 History 为 KEEP_LAST 时，保留的消息数量。

4. 持久性（Durability）

   • VOLATILE（易失）：发布者发送的消息只对当前在线的订阅者有效。默认。
   • TRANSIENT_LOCAL（瞬态本地）：发布者会保留最后一条消息，新订阅者加入时会收到这条消息。适用于状态发布。

QoS 不匹配问题

如果发布者和订阅者的 QoS 策略不兼容，它们将无法通信。这是 ROS2 开发中最常见的问题之一。

常见场景的 QoS 配置建议：

场景	可靠性	历史	深度	持久性
传感器数据（摄像头、激光雷达）	BEST_EFFORT	KEEP_LAST	1	VOLATILE
控制指令	RELIABLE	KEEP_LAST	10	VOLATILE
状态发布	RELIABLE	KEEP_LAST	1	TRANSIENT_LOCAL
日志	RELIABLE	KEEP_ALL	-	VOLATILE

五、常见问题与最佳实践

1. 节点间无法通信排查步骤：

• 检查域 ID 是否相同：echo $ROS_DOMAIN_ID；
• 检查防火墙是否关闭：sudo ufw disable；
• 检查话题 / 服务 / 动作名称是否一致；
• 检查 QoS 策略是否匹配：ros2 topic info -v <topic_name>；
• 检查 DDS 实现是否兼容：ROS2 默认使用 Fast DDS；优先使用话题传输高频数据流；

2. 最佳实践：

• 优先使用话题传输高频数据流；
• 使用服务处理一次性请求，避免耗时操作；
• 必须使用动作处理任何需要超过 1 秒的任务；
• 为不同类型的数据配置合适的 QoS 策略；
• 避免使用全局话题，合理使用命名空间；
• 多机通信时，确保所有机器在同一局域网且域 ID 相同；