ROS2核心概念与架构详解：从零开始机器人操作系统（1）

一、顶级架构一句话总结

节点 → DDS通信 → 话题/服务/动作 → 参数 → 工具链 → 机器人应用

ROS2（Robot Operating System 2）是新一代开源机器人操作系统，采用DDS作为通信中间件，去掉了ROS1的Master节点，提供更好的实时性和安全性。

二、ROS2 vs ROS1核心对比

1. 主要区别

2. ROS2的优势

• 去中心化架构：没有Master节点，任何节点故障不影响其它节点。

• DDS通信：工业级通信标准，支持QoS（服务质量）配置。

• 实时性：支持硬实时操作系统（RTOS）。

• 安全性：支持DDS-Security，提供加密和认证。

• 多语言支持：C++、Python、Java等。

三、ROS2核心概念

1. 节点（Node）

节点是ROS2的基本执行单元，每个节点负责一个特定功能。

#!/usr/bin/env python3 import rclpy from rclpy.node import Node class MinimalNode(Node): def __init__(self): super().__init__('minimal_node') self.get_logger().info('ROS2节点已启动！') def main(args=None): rclpy.init(args=args) node = MinimalNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

2. 话题（Topic）

话题是节点间的单向异步通信方式，发布-订阅模式。

3. 服务（Service）

服务是节点间的双向同步通信方式，请求-响应模式。

4. 动作（Action）

动作是节点间的双向异步通信方式，支持反馈和取消。

5. 参数（Parameter）

参数是节点的配置选项，支持运行时修改。

class ParamNode(Node): def __init__(self): super().__init__('param_node') # 声明参数 self.declare_parameter('robot_name', 'my_robot') self.declare_parameter('max_speed', 1.0) # 获取参数 robot_name = self.get_parameter('robot_name').value max_speed = self.get_parameter('max_speed').value self.get_logger().info(f'机器人名称: {robot_name}') self.get_logger().info(f'最大速度: {max_speed}')

四、ROS2架构详解

1. 中间件层（Middleware）

ROS2使用DDS作为通信中间件：

┌─────────────────────────────────────────┐
│ 用户应用层（Nodes） │
├─────────────────────────────────────────┤
│ ROS Client Library (rcl) │
├─────────────────────────────────────────┤
│ ROS Middleware (rmw) │
├─────────────────────────────────────────┤
│ DDS Implementation │
│ (FastDDS, CycloneDDS, RTI Connext) │
├─────────────────────────────────────────┤
│ 网络传输层 │
└─────────────────────────────────────────┘

2. DDS（Data Distribution Service）

DDS是OMG（对象管理组织）定义的工业级通信标准：

• DCPS（Data-Centric Publish-Subscribe）：以数据为中心的发布订阅。

• DDSI-RTPS：DDS互操作协议。

• QoS（Quality of Service）：服务质量策略。

3. QoS策略（服务质量）

ROS2提供多种QoS策略配置：

from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy # 定义QoS配置 qos_profile = QoSProfile( reliability=ReliabilityPolicy.BEST_EFFORT, durability=DurabilityPolicy.VOLATILE, depth=10 ) # 使用QoS创建发布者 self.publisher = self.create_publisher( String, 'sensor_data', qos_profile )

更多内容请看下回。