跳转至

第 3 章 工程化工具链

launch 系统负责将多个节点按拓扑序组织起来并注入参数,rosidl 负责将接口定义编译为各语言的类型代码,ros2 bag 负责离线数据录制与回放。这三样工具是从"一个节点能跑"到"一整套系统可复现地启动"的桥梁。本章给出 launch 文件的编写模式、接口包的生成链、组件化的进程模型,以及 bag 录制回放的常用操作——覆盖工程化落地的全部胶水环节。

1. launch 分层与 Entity/Action 模型

ROS 2 launch 系统由三层组成:

层 包名 职责
核心引擎 launch 描述语言无关的执行模型:Entity、Action、Condition、事件
ROS 扩展 launch_ros ROS 特化:Node、LifecycleNode、Parameter、remap 等
前端格式 Python / XML / YAML 用户编写的 launch 文件入口

Entity 是 launch 文件中的声明式对象(Node、Include 等),描述"做什么";Action 是运行时执行的命令(TimerAction、StartProcess、EmitEvent),描述"怎么做"。Entity 被 describe 为 Action 后由引擎执行。

Python launch 文件的骨架:

"""my_bringup.launch.py — 启动 talker + listener 并注入参数"""
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import Node

def generate_launch_description():
    # 声明可从命令行覆盖的 launch 参数
    publish_topic_arg = DeclareLaunchArgument(
        'publish_topic', default_value='/my_chatter',
        description='话题 remap 目标名')

    # demo_nodes_cpp 的 talker 没有发布频率参数(周期硬编码 1s,Python 版演示节点才有 publish_rate),
    # remap 与参数统一经 --ros-args 传入,LaunchConfiguration 在此消费
    talker = Node(
        package='demo_nodes_cpp',
        executable='talker',
        name='my_talker',
        arguments=[
            '--ros-args',
            '-r', ['chatter:=', LaunchConfiguration('publish_topic')],  # remap:chatter 换名
            '--param', 'use_sim_time:=false',                            # 节点参数注入
        ],
    )

    listener = Node(
        package='demo_nodes_cpp',
        executable='listener',
        name='my_listener',
        arguments=['--ros-args', '-r', ['chatter:=', LaunchConfiguration('publish_topic')]],
    )

    return LaunchDescription([publish_topic_arg, talker, listener])

启动方式:

ros2 launch my_bringup my_bringup.launch.py publish_topic:=/my_chatter

命令行传入的 publish_topic:=/my_chatter 覆盖了 DeclareLaunchArgument 的默认值。

2. DeclareLaunchArgument 与条件启动

DeclareLaunchArgument 在 launch 文件内声明可配置项。配合 IfCondition / UnlessCondition 实现条件启动:

from launch.conditions import IfCondition

debug_node = Node(
    package='my_pkg', executable='debug_tool',
    condition=IfCondition(LaunchConfiguration('enable_debug')),
)
ros2 launch my_pkg my.launch.py enable_debug:=true   # debug_node 才会启动

IfCondition 的判断依据是字符串求值结果:true/True/1 为真,其余为假。布尔型参数在 shell 与 YAML 中的表示不一致——true(小写)是 launch 条件判断的标准写法。

3. IncludeLaunchDescription 与 GroupAction

大型系统用多个 launch 文件组合。Include 引入另一个 launch 文件,GroupAction 给一组节点统一设置命名空间与条件:

from launch.actions import IncludeLaunchDescription, GroupAction
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch_ros.actions import Node, PushRosNamespace
from ament_index_python.packages import get_package_share_directory
import os

## Include:引入另一个包的 launch 文件(被包含文件必须真实存在)
sensors = IncludeLaunchDescription(
    PythonLaunchDescriptionSource(
        os.path.join(get_package_share_directory('my_pkg'), 'launch', 'sensors.launch.py')
    ),
    launch_arguments={'use_sim': 'false'}.items(),
)

## 启动 RViz2 并加载指定配置:rviz2 没有可 include 的 launch 文件,直接起节点
rviz = Node(
    package='rviz2',
    executable='rviz2',
    arguments=['-d', my_config_path],
)

## 一组节点共用命名空间
robot_group = GroupAction([
    PushRosNamespace('robot_1'),           # 组内所有节点加上 /robot_1 前缀
    Node(package='my_pkg', executable='sensor_driver'),
    Node(package='my_pkg', executable='controller'),
])

get_package_share_directory 解析已安装包的 share/ 路径,是跨包引用 launch 文件的标准方式。

4. 事件驱动

launch 引擎支持基于事件的流程控制。常用场景:某进程退出后触发重启、超时后强制终止。

from launch.actions import RegisterEventHandler, TimerAction
from launch.event_handlers import OnProcessExit
from launch.events import Shutdown

## sensor_driver 退出 2 秒后自动重启
restart_after_exit = RegisterEventHandler(
    OnProcessExit(
        target_action=sensor_node,
        on_exit=[TimerAction(period=2.0, actions=[sensor_node])],
    )
)

## 延迟 5 秒再启动 controller(等传感器初始化完成)
delayed_controller = TimerAction(period=5.0, actions=[controller_node])

事件模型是异步的——RegisterEventHandler 注册监听器,事件触发时回调执行对应 Action。这是实现"有序启动"最轻量的方式,无需 lifecycle 状态机介入。

5. rosidl 生成链与接口包

.msg、.srv、.action 文件经 rosidl 工具链编译为各语言的类型定义:

flowchart LR
    idl[".msg / .srv / .action"] --> ad["rosidl_adapter<br/>解析为 IDL"]
    ad --> gpy["rosidl_generator_py<br/>Python 类"]
    ad --> gcpp["rosidl_generator_cpp<br/>C++ 头文件"]
    gpy --> ts["rosidl_typesupport_fastrtps_c/cpp<br/>序列化支持"]
    gcpp --> ts

接口包必须是 ament_cmake 类型。最小接口包结构:

my_interfaces/
├── CMakeLists.txt
├── package.xml
├── msg/
│   └── MotorState.msg        # float64 position / float64 velocity / float64 effort
├── srv/
│   └── SetMode.srv           # std_msgs/String mode → bool success
└── action/
    └── RotateTo.action       # float64 target_angle → float64 result_angle / 反馈: float64 current_angle

CMakeLists.txt:

cmake_minimum_required(VERSION 3.8)
project(my_interfaces)
find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
find_package(std_msgs REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/MotorState.msg"
  "srv/SetMode.srv"
  "action/RotateTo.action"
  DEPENDENCIES std_msgs          # 接口内引用了 std_msgs 类型时声明
)
ament_export_dependencies(rosidl_default_runtime)
ament_package()

package.xml 中须声明三行关键依赖:

<buildtool_depend>rosidl_default_generators</buildtool_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>  <!-- 标记为接口包 -->

接口文件语法——MotorState.msg:

float64 position      # 关节角度 (rad)
float64 velocity      # 角速度 (rad/s)
float64 effort        # 力矩 (N·m)

.srv 文件用 --- 分隔请求与响应;.action 文件用两个 --- 分隔目标、结果、反馈三段。

生成后可在 Python 中导入:from my_interfaces.msg import MotorState。消息不支持继承——用组合(嵌套已定义的消息类型)替代。常量用 uint8 STATUS_IDLE=0 语法声明,默认值在字段后追加(如 float64 kp 1.0)。

6. 组件化(composable nodes)

组件是把多个节点装进同一个进程运行的机制——共享 DDS participant,省去进程间通信的序列化开销,适合高频数据通路。这个机制只对 C++ 节点成立:节点要编译成 rclcpp_components 插件才能被容器加载,Python 节点没有对应的注册与加载路径,多节点只能靠同一 executor(第 2 章「一名多节点与组件化边界」)。

from launch_ros.descriptions import ComposableNode
from launch_ros.actions import ComposableNodeContainer, LoadComposableNodes

## 创建 component container(一个空壳进程,也可用 load-components 参数启动时直接载入)
container = ComposableNodeContainer(
    name='my_container',
    namespace='',
    package='rclcpp_components',
    executable='component_container',
    composable_node_descriptions=[
        ComposableNode(
            package='my_components',
            plugin='my_components::SensorNode',
            name='sensor_node',
            parameters=[{'rate': 100.0}],
        ),
    ],
    output='screen',
)

## 或者:容器已在运行,运行期通过 service 调用动态加载组件
load_composable = LoadComposableNodes(
    target_container='my_container',
    composable_descriptions=[
        ComposableNode(
            package='my_components',
            plugin='my_components::ControllerNode',
            name='controller_node',
        ),
    ],
)

组件化 vs 独立进程的取舍:

维度 独立进程 组件化
崩溃隔离 单节点崩溃不影响其他 一个组件异常可能拖垮整个 container
通信开销 跨进程走 DDS 序列化 同进程内共享内存,零拷贝
调试 每进程独立 GDB/日志 多组件混在同一进程日志
适用 低频节点、异构语言 高频传感器融合、 tightly coupled 控制链

7. YAML 参数文件

多参数场景用 YAML 文件批量注入,避免命令行堆积:

## robot_params.yaml
/**:                              # 通配所有节点
  ros__parameters:
    use_sim_time: false

sensor_driver:                    # 节点名匹配
  ros__parameters:
    frame_id: "laser_link"
    scan_frequency: 10.0          # Hz
    quality_filter: [1, 2, 3]     # 列表

controller_node:
  ros__parameters:
    pid:
      kp: 1.5
      ki: 0.01
      kd: 0.1

加载方式:

ros2 run my_pkg sensor_driver --ros-args --params-file robot_params.yaml

缩进坑:ros__parameters 必须作为节点名的直接子键,多一级缩进会导致参数被静默忽略。/** 匹配任意命名空间下的所有节点;/robot_1/** 只匹配 /robot_1 命名空间下的节点。YAML 中 true/false 是布尔,"true" 是字符串——launch 的 IfCondition 只认字符串 true。

坑:YAML 层级写错,参数静默不生效

节点参数必须落在 /**: ros__parameters: 之下,缩进错一层就变成“另一个节点”的配置;启动后按 ros2 param dump <node> 核实,而不是只看文件写过。

8. ros2 CLI 与排障自检

常用命令行工具速查:

## 节点
ros2 node list / info <node_name>

## 话题
ros2 topic list -t                        # 带类型
ros2 topic echo /scan                     # 打印消息内容
ros2 topic hz /scan                       # 频率统计
ros2 topic bw /camera/image_raw           # 带宽 (bytes/s)
ros2 topic info /scan --verbose           # 显示 pub/sub 数量与 QoS

## 服务
ros2 service list
ros2 service call /set_mode my_interfaces/srv/SetMode "{mode: {data: 'auto'}}"

## 参数
ros2 param list /my_node
ros2 param get /my_node max_vel
ros2 param set /my_node max_vel 1.5
ros2 param dump /my_node                  # 导出当前节点全部参数

## 动作
ros2 action list
ros2 action send_goal /navigate my_interfaces/action/RotateTo "{target_angle: 1.57}"

日志级别控制:

## 运行时修改节点日志级别
ros2 service call /my_node/set_logger_level rcl_interfaces/srv/SetLoggerLevel \
  "{logger: 'my_node', level: 10}"   # 10=DEBUG 20=INFO 30=WARN 40=ERROR 50=FATAL

ROS_LOG_DIR 环境变量控制日志输出目录。默认日志以 session 为单位存于 ~/.ros/log/。

8.1 按问题表现逐层排查

现象 排查步骤
ros2 topic list 看不到目标话题 1. 确认 ROS_DOMAIN_ID 一致;2. ros2 daemon stop && start 刷新缓存;3. 确认节点已启动(ros2 node list)
话题存在但 echo 无输出 1. ros2 topic info --verbose 检查 QoS 兼容性;2. 确认消息类型一致(-t 参数);3. 检查 remap 是否导致订阅端连到了另一个话题
跨机无法通信 1. 防火墙是否放行 DDS 端口(Fast DDS 默认 7400+domain*250);2. 组播是否可达(ping -I <iface> 239.255.0.1);3. 两端 RMW_IMPLEMENTATION 一致
服务调用无响应 1. ros2 service find 确认服务端存在;2. 接口类型与 .srv 定义匹配;3. executor 死锁——服务端回调是否阻塞(见第 2 章 executor)
bag 录制后无数据 1. 录制时话题是否真的有消息(ros2 topic hz 确认);2. 存储格式兼容性(mcap 需要插件)

8.2 上电即用诊断清单

系统上电、环境 source 完成后,启动业务节点前按序过一遍——前三步不依赖 DDS 网络,启动异常时也能执行:

env | grep ROS          # 1. ROS_DISTRO / ROS_LOCALHOST_ONLY / ROS_DOMAIN_ID / RMW_IMPLEMENTATION 是否符合预期
ros2 doctor             # 2. 环境完整性:无 WARNING/ERROR 才算过(第 1 章)
ros2 pkg prefix demo_nodes_cpp   # 3. 包可定位:能解析出安装路径即环境加载链正常
ros2 daemon stop && ros2 daemon start  # 4. CLI 缓存重置:列表类命令查询异常的第一处置
ros2 topic list         # 5. 发现面:能看到预期话题即 DDS 组播与域隔离正常

顺序背后的机制:第 1–3 步检查的是静态环境(变量、包索引、hook 加载链);第 4 步处理 daemon 缓存过期——CLI 查询由 daemon 代理,不代表真实网络状态;第 5 步才进入动态发现面,此时看不到话题应回到第 2 章的域隔离与 rmw 一致性排查。

判据:上电自检按三件走

ros2 doctor 看环境、ros2 daemon status 看发现服务、再确认 ROS_DISTRO 与工作空间 overlay 已加载——三步过不了,后面所有命令都不可信。

9. ros2 bag 录制与回放

ros2 bag 是离线数据分析的基础工具,录制选定话题的时间序列、回放时重新注入图中。

## 录制全部话题(生产环境建议只录所需话题,减少磁盘压力)
ros2 bag record -a

## 只录传感器与控制话题
ros2 bag record /scan /odom /cmd_vel -o my_experiment --storage mcap

## 回放(仿真模式下需 use_sim_time)
ros2 bag play my_experiment --clock

存储格式:Humble 默认 sqlite3,mcap 格式压缩率更高且支持 zstd。安装 mcap 插件:

sudo apt install -y ros-humble-mcap-ros2-support

常用分析操作:

ros2 bag info my_experiment       # 查看话题、时长、消息数

提取部分话题:Humble 的 CLI 没有 ros2 bag filter 子命令(ros2 bag 仅有 convert / info / list / play / record / reindex),话题筛选在录制时用 --topics 前置完成,或事后用 ros2 bag convert 配合输出配置 YAML 实现(输出项的 topics 字段列出保留话题)。

bag 文件目录结构:一个 .db3(或 .mcap)+ 一个 metadata.yaml。metadata.yaml 记录话题列表、类型、起止时间——损坏后 bag 不可识别。

10. 要点

  • launch 系统三层结构(launch / launch_ros / 前端格式)将多节点编排、参数注入与条件启动统一为一个文件;Python 格式功能最完整。
  • 接口包必须是 ament_cmake 类型,rosidl_generate_interfaces 是唯一的接口生成入口;msg 不支持继承,用组合与嵌套。
  • 组件化把多个节点装入同一进程省序列化开销,代价是崩溃隔离减弱——高频数据通路用组件,异构语言和独立调试场景用进程。
  • YAML 参数文件中 ros__parameters 的缩进层级是最常触发的静默失败点。
  • ros2 bag + mcap 组合是离线分析的数据源基础;录制时指定话题比 -a 更可控。
  • 通信排障按"可见性 → QoS 兼容性 → 类型一致性 → executor 阻塞"四层递进,覆盖绝大多数"连不上/收不到"的场景;上电后用五步自检(环境变量 → doctor → 包定位 → daemon 重置 → 话题可见)把静态环境与动态发现面分开确认。

参考文献与延伸阅读

  • colcon. 构建工具仓库与文档. 仓库
  • ROS 2. Rosdep 教程. 文档
  • ROS 2. Humble 文档. 文档