ROS2 机器人不是"跑个demo"：QoS、生命周期、动作、实时性与多机部署的工程底线，附可运行代码

# qos_profiles.py
from rclpy.qos import (QoSProfile, ReliabilityPolicy, HistoryPolicy,
                       DurabilityPolicy, LivelinessPolicy)

# 传感器：丢旧留新，尽力而为
SENSOR_QOS = QoSProfile(
    reliability=ReliabilityPolicy.BEST_EFFORT,
    history=HistoryPolicy.KEEP_LAST, depth=1,
    durability=DurabilityPolicy.VOLATILE,
    liveliness=LivelinessPolicy.AUTOMATIC,
    deadline=None, lifespan=None)

# 控制/状态：可靠、保序、有深度
CTRL_QOS = QoSProfile(
    reliability=ReliabilityPolicy.RELIABLE,
    history=HistoryPolicy.KEEP_LAST, depth=10,
    durability=DurabilityPolicy.VOLATILE)

# 参数/地图：RELIABLE + TRANSIENT_LOCAL，晚加入也能拿到
LATCHED_QOS = QoSProfile(
    reliability=ReliabilityPolicy.RELIABLE,
    history=HistoryPolicy.KEEP_LAST, depth=1,
    durability=DurabilityPolicy.TRANSIENT_LOCAL)

# motor_driver.py
import rclpy
from rclpy.lifecycle import LifecycleNode, TransitionCallbackReturn

class MotorDriver(LifecycleNode):
    def __init__(self):
        super().__init__('motor_driver')
        self.pub = None

    def on_configure(self, state):
        self.declare_parameter('can_interface', 'can0')
        iface = self.get_parameter('can_interface').value
        if not self._open_can(iface):          # 硬件自检
            return TransitionCallbackReturn.FAILURE
        self.pub = self.create_lifecycle_publisher(
            std_msgs.msg.Float64, '/cmd_vel', 10)
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state):
        self.get_logger().info('电机驱动已激活，允许下发速度')
        return super().on_activate(state)

    def on_deactivate(self, state):
        self._safe_stop()                      # 去激活必须先安全停车
        return super().on_deactivate(state)

    def on_cleanup(self, state):
        self._close_can()
        return TransitionCallbackReturn.SUCCESS

# navigate_server.py
import rclpy
from rclpy.action import ActionServer, CancelResponse, GoalResponse
from rclpy.action.server import ServerGoalHandle
from nav2_msgs.action import NavigateToPose

class NavServer(rclpy.node.Node):
    def __init__(self):
        super().__init__('nav_server')
        self._srv = ActionServer(
            self, NavigateToPose, 'navigate_to_pose',
            execute_callback=self.execute,
            goal_callback=self.on_goal,
            cancel_callback=lambda _: CancelResponse.ACCEPT,
            handle_accepted_callback=self.on_accepted)

    def on_goal(self, req):
        if self._battery_low():
            return GoalResponse.REJECT      # 业务前置校验
        return GoalResponse.ACCEPT

    def execute(self, gh: ServerGoalHandle):
        fb = NavigateToPose.Feedback()
        for step in self._plan_and_drive(gh.request.goal):
            if gh.is_cancel_requested:
                gh.canceled(); return NavigateToPose.Result(success=False)
            fb.distance_remaining = step.remain
            gh.publish_feedback(fb)          # 进度反馈
        gh.succeed()
        return NavigateToPose.Result(success=True)

# main.py
import rclpy
from rclpy.executors import MultiThreadedExecutor
from rclpy.callback_groups import ReentrantCallbackGroup, MutuallyExclusiveCallbackGroup

ctrl_group = MutuallyExclusiveCallbackGroup()   # 控制：互斥，保序
sensor_group = ReentrantCallbackGroup()          # 传感器：可并发

node.create_subscription(Twist, '/cmd_vel', ctrl_cb, 10, callback_group=ctrl_group)
node.create_subscription(Image, '/camera', img_cb, 1, callback_group=sensor_group)

executor = MultiThreadedExecutor(num_threads=4)
executor.add_node(node); executor.spin()

// zero_copy_pub.cpp（rclcpp片段）
auto loaned = publisher_->borrow_loaned_message();
fill_pointcloud(loaned.get());              // 直接写共享内存
publisher_->publish(std::move(loaned));     // 无拷贝发布

export ROS_DOMAIN_ID=42          # 不同机器人/租户不同域
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
ros2 security create_enclave /path/keystore /robot1
export ROS_SECURITY_ENCLAVE=/robot1 ROS_SECURITY_ENABLE=true