第 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 重置 → 话题可见)把静态环境与动态发现面分开确认。