【ROS/ROS2与实时Linux系列】第二十八篇 ROS 2 `launch`系统：实时任务启动编排

一、简介：为什么 launch 系统需要结合实时 Linux？

在 ROS 2 机器人系统中，节点数量往往从几十个到上百个不等，涵盖传感器驱动、感知算法、规划控制、通信中间件等多个层级。传统的 launch 文件只能保证"启动起来"，但在实时场景下存在致命缺陷：

• 启动顺序混乱：导航节点先于定位节点启动，导致初始路径规划错误

• 优先级不分：视觉处理节点与电机控制节点平级竞争 CPU，控制指令延迟抖动

• 资源隔离缺失：关键任务被系统后台进程抢占，实时性无法保障

掌握 ROS 2 launch 系统的扩展能力，结合实时 Linux 的调度策略，可以实现：

• 确定性启动顺序：按依赖关系分层启动，确保数据流正确

• 精细化资源分配：为关键节点绑定 CPU 核心、设置实时优先级

• 运行时动态调整：根据系统负载动态迁移任务，优化整体性能

这是构建工业级、车规级 ROS 2 系统的必备技能。

二、核心概念：launch 系统与实时调度术语

三、环境准备：搭建实时 ROS 2 开发环境

3.1 硬件需求

3.2 软件环境

3.3 一键安装实时内核与 ROS 2

#!/bin/bash # install_rt_ros2.sh - 一键安装环境 set -e echo "=== 安装 ROS 2 Humble ===" sudo apt update && sudo apt install -y curl gnupg lsb-release sudo curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.key -o /usr/share/keyrings/ros-archive-keyring.gpg echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] http://packages.ros.org/ros2/ubuntu jammy main" | sudo tee /etc/apt/sources.list.d/ros2.list > /dev/null sudo apt update sudo apt install -y ros-humble-desktop ros-humble-ros2-control ros-humble-ros2-controllers echo "=== 安装实时内核 5.15.0-rt ===" sudo apt install -y linux-image-5.15.0-rt linux-headers-5.15.0-rt echo "=== 安装实时工具 ===" sudo apt install -y rt-tests util-linux echo "=== 配置环境 ===" echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc echo "export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp" >> ~/.bashrc echo "=== 请重启选择 RT 内核 ==="

3.4 验证实时环境

# 验证内核 uname -r # 应包含 "rt" # 验证实时调度可用 chrt --max # 显示最大优先级 99 # 测试调度延迟 sudo cyclictest -p99 -i100 -d60s -n # Max latency 应 < 50μs

四、应用场景：自动驾驶车辆的多传感器融合控制

在自动驾驶车辆中，ROS 2 节点通常包括：

挑战：传统 launch 文件同时启动所有节点，控制节点可能被感知节点的数据洪流淹没，导致控制指令延迟，车辆抖动。

解决方案：使用扩展 launch 系统，实现：

1. 分层启动：先启动控制层，确保车辆安全；再启动感知层；最后启动规划层

2. 优先级分级：控制节点 SCHED_FIFO 99，感知节点 SCHED_FIFO 80，规划节点 CFS

3. CPU 隔离：控制节点独占 CPU 0，其他节点使用 CPU 1-3

4. 动态监控：节点崩溃自动重启，负载过高时迁移非关键任务

五、实际案例与步骤：构建实时感知-控制 launch 系统

5.1 基础 launch 文件结构

创建功能包：

cd ~/ros2_ws/src ros2 pkg create --build-type ament_python rt_launch_demo --dependencies launch_ros cd rt_launch_demo mkdir -p launch config

5.2 阶段一：静态优先级配置（入门级）

文件：launch/static_priority.launch.py

#!/usr/bin/env python3 """ 静态优先级配置 launch 文件 为关键节点设置 SCHED_FIFO 优先级 """ from launch import LaunchDescription from launch_ros.actions import Node from launch.actions import ExecuteProcess from launch.substitutions import FindExecutable def generate_launch_description(): ld = LaunchDescription() # 控制节点：最高优先级，硬实时 # 使用 taskset 绑定 CPU 0，chrt 设置 SCHED_FIFO 99 control_node = ExecuteProcess( cmd=[ 'taskset', '-c', '0', # 绑定 CPU 0 'chrt', '-f', '99', # SCHED_FIFO 优先级 99 'ros2', 'run', 'controller_manager', 'ros2_control_node', '--ros-args', '--params-file', '/config/controller.yaml' ], name='control_node', output='screen', shell=False ) ld.add_action(control_node) # 激光雷达驱动：高优先级，绑定 CPU 1 lidar_node = ExecuteProcess( cmd=[ 'taskset', '-c', '1', 'chrt', '-f', '80', 'ros2', 'run', 'velodyne_driver', 'velodyne_driver_node', '--ros-args', '-p', 'device_ip:=192.168.1.201' ], name='lidar_node', output='screen' ) ld.add_action(lidar_node) # 规划节点：普通优先级，CFS 调度 planner_node = Node( package='nav2_planner', executable='planner_server', name='planner_server', parameters=['/config/nav2_params.yaml'], # 不设置 chrt，使用默认 CFS ) ld.add_action(planner_node) return ld

使用场景：快速验证实时配置，适合节点数量少、配置固定的场景。

5.3 阶段二：动态参数与条件启动（进阶级）

文件：launch/dynamic_priority.launch.py

#!/usr/bin/env python3 """ 动态优先级配置 launch 文件 支持命令行参数传入优先级，条件启动不同配置 """ from launch import LaunchDescription, LaunchContext from launch_ros.actions import Node from launch.actions import ExecuteProcess, DeclareLaunchArgument, OpaqueFunction from launch.substitutions import LaunchConfiguration, PythonExpression def evaluate_priority(context: LaunchContext, *args, **kwargs): """ OpaqueFunction: 在 launch 解析期执行 Python 代码 根据系统负载动态计算优先级 """ priority = int(context.launch_configurations.get('base_priority', '50')) # 读取 CPU 负载 import os load_avg = os.getloadavg()[0] # 1分钟平均负载 # 负载高时降低非关键任务优先级 if load_avg > 4.0: planner_priority = max(1, priority - 20) print(f"高负载 detected: planner 优先级降至 {planner_priority}") else: planner_priority = priority return [str(planner_priority)] def generate_launch_description(): ld = LaunchDescription() # 声明启动参数 ld.add_action(DeclareLaunchArgument( 'base_priority', default_value='50', description='基础优先级 (1-99)' )) ld.add_action(DeclareLaunchArgument( 'enable_realtime', default_value='true', description='是否启用实时调度' )) # 条件表达式：根据 enable_realtime 选择启动方式 enable_rt = LaunchConfiguration('enable_realtime') # 控制节点：条件启动 # 启用实时时使用 chrt，否则普通启动 control_cmd = [ 'ros2', 'run', 'controller_manager', 'ros2_control_node' ] control_rt_cmd = ['taskset', '-c', '0', 'chrt', '-f', '99'] + control_cmd # 使用 PythonExpression 实现条件选择 control_node = ExecuteProcess( cmd=PythonExpression([ "['taskset', '-c', '0', 'chrt', '-f', '99'] + ", "['ros2', 'run', 'controller_manager', 'ros2_control_node'] ", "if '", enable_rt, "' == 'true' else ", "['ros2', 'run', 'controller_manager', 'ros2_control_node']" ]), shell=True # 需要 shell 解析表达式 ) # 简化：使用两个独立的 ExecuteProcess 配合 IfCondition # 更清晰的写法：使用 OpaqueFunction 完全自定义 def launch_control(context, *args, **kwargs): use_rt = context.launch_configurations.get('enable_realtime', 'true') == 'true' if use_rt: return [ExecuteProcess( cmd=['taskset', '-c', '0', 'chrt', '-f', '99', 'ros2', 'run', 'controller_manager', 'ros2_control_node'], output='screen' )] else: return [Node( package='controller_manager', executable='ros2_control_node', output='screen' )] ld.add_action(OpaqueFunction(function=launch_control)) # 规划节点：动态优先级 def launch_planner(context, *args, **kwargs): base = int(context.launch_configurations.get('base_priority', '50')) # 规划节点优先级 = 基础 - 10 planner_prio = max(1, base - 10) return [ExecuteProcess( cmd=['chrt', '-f', str(planner_prio), 'ros2', 'run', 'nav2_planner', 'planner_server'], output='screen' )] ld.add_action(OpaqueFunction(function=launch_planner)) return ld

运行方式：

# 默认配置 ros2 launch rt_launch_demo dynamic_priority.launch.py # 指定基础优先级，关闭实时 ros2 launch rt_launch_demo dynamic_priority.launch.py base_priority:=30 enable_realtime:=false

5.4 阶段三：完整实时编排系统（生产级）

文件：launch/rt_orchestrator.launch.py

#!/usr/bin/env python3 """ 生产级实时任务编排系统 特性：分层启动、资源隔离、健康监控、自动恢复 """ from launch import LaunchDescription, LaunchContext, SomeSubstitutionsType from launch_ros.actions import Node from launch.actions import ( ExecuteProcess, DeclareLaunchArgument, OpaqueFunction, GroupAction, TimerAction, RegisterEventHandler ) from launch.event_handlers import OnProcessExit, OnProcessStart from launch.substitutions import LaunchConfiguration, PythonExpression from launch.conditions import IfCondition, UnlessCondition import os import yaml # 节点配置模板 RT_NODE_CONFIG = { 'control': { 'cpu': '0', 'priority': 99, 'policy': 'fifo', # fifo, rr, or cfs 'package': 'controller_manager', 'executable': 'ros2_control_node', 'startup_delay': 0.0, # 最先启动 'restart_on_failure': True }, 'lidar_driver': { 'cpu': '1', 'priority': 80, 'policy': 'fifo', 'package': 'velodyne_driver', 'executable': 'velodyne_driver_node', 'startup_delay': 1.0, # 控制启动后 1s 'restart_on_failure': True }, 'camera_driver': { 'cpu': '2', 'priority': 75, 'policy': 'fifo', 'package': 'usb_cam', 'executable': 'usb_cam_node_exe', 'startup_delay': 1.0, 'restart_on_failure': False # 相机故障不致命 }, 'slam': { 'cpu': '2-3', 'priority': 50, 'policy': 'rr', # 轮询，避免长时间占用 'package': 'slam_toolbox', 'executable': 'async_slam_toolbox_node', 'startup_delay': 5.0, # 传感器稳定后启动 'restart_on_failure': False }, 'planner': { 'cpu': '3', 'priority': 0, # CFS，非实时 'policy': 'cfs', 'package': 'nav2_planner', 'executable': 'planner_server', 'startup_delay': 6.0, 'restart_on_failure': False } } def build_rt_command(node_name: str, config: dict) -> list: """ 根据配置构建带实时调度的启动命令 """ cmd = [] # CPU 绑定 if config['cpu']: cmd.extend(['taskset', '-c', config['cpu']]) # 调度策略 if config['policy'] == 'fifo': cmd.extend(['chrt', '-f', str(config['priority'])]) elif config['policy'] == 'rr': cmd.extend(['chrt', '-r', str(config['priority'])]) # cfs: 不添加 chrt，使用默认 CFS # ROS 2 节点 cmd.extend([ 'ros2', 'run', config['package'], config['executable'], '--ros-args', '-r', f'__node:={node_name}_rt' ]) return cmd def create_monitored_process(node_name: str, config: dict): """ 创建带监控的进程，支持自动重启 """ cmd = build_rt_command(node_name, config) process = ExecuteProcess( cmd=cmd, name=node_name, output='screen', respawn=config.get('restart_on_failure', False), respawn_delay=5.0 ) # 添加启动/退出日志 process.register_event_handler( OnProcessStart( on_start=[lambda ctx: print(f"[RT-ORCH] {node_name} started")] ) ) process.register_event_handler( OnProcessExit( on_exit=[lambda ctx: print(f"[RT-ORCH] {node_name} exited")] ) ) return process def generate_launch_description(): ld = LaunchDescription() # 全局参数 ld.add_action(DeclareLaunchArgument( 'config_file', default_value='', description='YAML 配置文件路径（可选）' )) ld.add_action(DeclareLaunchArgument( 'dry_run', default_value='false', description='仅打印命令，不实际启动' )) def orchestrate(context: LaunchContext, *args, **kwargs): """ 核心编排逻辑：按 startup_delay 排序，分层启动 """ config_path = context.launch_configurations.get('config_file', '') dry_run = context.launch_configurations.get('dry_run', 'false') == 'true' # 加载外部配置（如果提供） config = RT_NODE_CONFIG.copy() if config_path and os.path.exists(config_path): with open(config_path) as f: config.update(yaml.safe_load(f)) # 按 startup_delay 分组 delay_groups = {} for name, cfg in config.items(): delay = cfg.get('startup_delay', 0.0) delay_groups.setdefault(delay, []).append((name, cfg)) actions = [] for delay in sorted(delay_groups.keys()): nodes = delay_groups[delay] for node_name, cfg in nodes: if dry_run: cmd = ' '.join(build_rt_command(node_name, cfg)) print(f"[DRY-RUN] delay={delay}s: {cmd}") continue process = create_monitored_process(node_name, cfg) if delay > 0: # 使用 TimerAction 延迟启动 actions.append(TimerAction( period=delay, actions=[process] )) else: actions.append(process) return actions ld.add_action(OpaqueFunction(function=orchestrate)) return ld

配置文件示例：config/override.yaml

# 覆盖默认配置，适应不同硬件 control: cpu: '0,4' # 绑定到 NUMA node 0 的核心 0 和 4 priority: 99 lidar_driver: cpu: '1' priority: 85 # 比默认更高 camera_driver: startup_delay: 2.0 # 延迟更久，等待 USB 稳定

运行：

# 默认配置 ros2 launch rt_launch_demo rt_orchestrator.launch.py # 使用外部配置 ros2 launch rt_launch_demo rt_orchestrator.launch.py config_file:=$(pwd)/config/override.yaml # 仅预览命令 ros2 launch rt_launch_demo rt_orchestrator.launch.py dry_run:=true

5.5 阶段四：运行时动态调整（高级）

文件：scripts/runtime_tuner.py

#!/usr/bin/env python3 """ 运行时动态调整节点优先级 订阅系统负载，自动迁移非关键任务 """ import os import psutil import subprocess import rclpy from rclpy.node import Node from std_msgs.msg import Float32 class RuntimeTuner(Node): def __init__(self): super().__init__('runtime_tuner') self.load_threshold = 6.0 # 1分钟负载阈值 self.migration_target = '3' # 非关键任务迁移目标 CPU self.timer = self.create_timer(5.0, self.check_and_tune) self.get_logger().info('Runtime tuner started') def check_and_tune(self): # 读取系统负载 load1, _, _ = os.getloadavg() cpu_count = os.cpu_count() self.get_logger().info(f'System load: {load1:.2f} / {cpu_count} CPUs') if load1 > self.load_threshold: self.get_logger().warn('High load detected, migrating non-critical tasks') self.migrate_non_critical() def migrate_non_critical(self): """ 将非关键任务迁移到指定 CPU 关键任务（控制、传感器）保持不动 """ # 查找 SLAM、规划等可迁移节点 for proc in psutil.process_iter(['pid', 'name', 'cmdline']): try: cmdline = ' '.join(proc.info['cmdline'] or []) # 匹配可迁移节点 if any(x in cmdline for x in ['slam_toolbox', 'planner_server', 'navigator']): pid = proc.info['pid'] # 使用 taskset 迁移 result = subprocess.run( ['taskset', '-pc', self.migration_target, str(pid)], capture_output=True, text=True ) if result.returncode == 0: self.get_logger().info( f'Migrated {proc.info["name"]}({pid}) to CPU {self.migration_target}' ) else: self.get_logger().error(f'Migration failed: {result.stderr}') except (psutil.NoSuchProcess, psutil.AccessDenied): continue def main(): rclpy.init() node = RuntimeTuner() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()

编译运行：

# 添加到 launch 中作为最后一个节点 # 或在运行时单独启动 ros2 run rt_launch_demo runtime_tuner

六、常见问题与解答

七、实践建议与最佳实践

7.1 调试技巧

# 查看节点实际调度策略 ps -eo pid,comm,cls,rtprio | grep ros2 # 实时监控 CPU 绑定 watch -n 1 'taskset -pc <pid>' # 测量调度延迟 sudo cyclictest -p99 -i100 -d300s -h400 -q > latency.hist

7.2 性能优化

• NUMA 感知：跨 NUMA 节点访问内存延迟高，控制节点绑定到与网卡同一 NUMA 节点

• 中断亲和性：将传感器网卡中断绑定到专用 CPU，避免与业务竞争

# 查看网卡中断 cat /proc/interrupts | grep eth # 绑定中断到 CPU 2 echo 2 > /proc/irq/123/smp_affinity

7.3 常见错误规避

• 优先级反转：关键节点使用PTHREAD_PRIO_INHERIT互斥锁

• 内存碎片：启动前预留大页内存echo 1024 > /proc/sys/vm/nr_hugepages

• 日志风暴：生产环境关闭 ROS 2 调试日志，使用RCUTILS_LOGGING_SEVERITY=WARN

八、总结与应用场景

通过本文的递进式讲解，我们构建了从简单到复杂的 ROS 2 实时任务编排系统：

核心要点：

• ExecuteProcess+taskset/chrt是实现实时调度的关键组合

• OpaqueFunction提供完全自定义能力，适合复杂逻辑

• 分层启动 + 延迟定时器确保依赖正确

• 运行时监控与自动恢复保障系统韧性

掌握这些技能，你就能构建出确定性强、资源隔离、故障自愈的 ROS 2 实时系统，满足自动驾驶、工业机器人、医疗设备等严苛场景的需求。立即将本文代码应用到你的项目中，体验从"能跑"到"跑得稳"的质变！