本文由 linuxROS 整理发布，首发于 linuxros.cn，转载请注明出处。

ROS2节点开发与生命周期：从最小节点到托管状态机

节点是ROS2最小执行单元，生命周期是节点行为可控的关键机制。基于Jazzy Jalisco LTS，从最简节点到生命周期状态机，从节点组合到命名空间隔离，所有代码可直接运行验证。

一、节点基础

ROS2中，节点（Node）是最小执行单元，每个节点是独立进程。一个机器人系统由多个节点协作完成：传感器节点采集数据，控制节点执行决策，处理节点跑算法，接口节点对接外部系统。

节点类型

类型	职责	典型示例
传感器节点	采集原始数据	camera_driver、lidar_driver、imu_driver
控制节点	执行运动决策	cmd_vel_controller、arm_controller
处理节点	运行算法逻辑	object_detector、slam_node、path_planner
接口节点	对接外部系统	joystick_adapter、web_bridge、mqtt_gateway

Python最简节点

import rclpy
from rclpy.node import Node

class MinimalNode(Node):
    def __init__(self):
        super().__init__('minimal_node')
        self.get_logger().info('节点已启动')

def main(args=None):
    rclpy.init(args=args)
    node = MinimalNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

运行验证：

# 假设包名为my_nodes，入口配置指向main
ros2 run my_nodes minimal_node
# 输出: [INFO] [minimal_node]: 节点已启动

这个节点什么也不做，但它包含了ROS2节点的完整骨架：继承Node、指定节点名、初始化/销毁/关闭三段式。

二、节点组成要素

一个实用节点不只是个空壳，它由六种核心要素组合而成：

要素	作用	API
发布者（Publisher）	向话题发送消息	`create_publisher()`
订阅者（Subscription）	从话题接收消息	`create_subscription()`
服务（Service）	提供请求-响应调用	`create_service()`
客户端（Client）	调用远程服务	`create_client()`
定时器（Timer）	周期执行回调	`create_timer()`
参数（Parameter）	动态配置键值对	`declare_parameter()`

节点内部结构

flowchart TB N["Node 节点核心"] --> P["Publisher 发布者"] N --> S["Subscription 订阅者"] N --> SV["Service 服务"] N --> C["Client 客户端"] N --> T["Timer 定时器"] N --> PM["Parameter 参数"] P -->|"发布"| T1["Topic"] S -->|"订阅"| T2["Topic"] SV -->|"响应"| R1["Request"] C -->|"请求"| R2["Service"] style N fill:#E3F2FD,stroke:#1976D2 style P fill:#E8F5E9,stroke:#388E3C style S fill:#E8F5E9,stroke:#388E3C style SV fill:#FFF8E1,stroke:#F57C00 style C fill:#FFF8E1,stroke:#F57C00 style T fill:#F3E5F5,stroke:#7B1FA2 style PM fill:#F3E5F5,stroke:#7B1FA2

Python示例：发布者+订阅者+定时器

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SensorProcessor(Node):
    def __init__(self):
        super().__init__('sensor_processor')

        # 参数
        self.declare_parameter('process_rate', 10.0)

        # 发布者：处理后的结果
        self.result_pub = self.create_publisher(String, 'processed_data', 10)

        # 订阅者：原始传感器数据
        self.raw_sub = self.create_subscription(
            String, 'raw_sensor_data', self.on_raw_data, 10)

        # 定时器：周期性状态报告
        rate = self.get_parameter('process_rate').value
        self.timer = self.create_timer(1.0 / rate, self.timer_callback)
        self.msg_count = 0

    def on_raw_data(self, msg: String):
        """收到原始数据，处理后发布"""
        result = String()
        result.data = f'已处理: {msg.data}'
        self.result_pub.publish(result)
        self.msg_count += 1

    def timer_callback(self):
        """定时报告处理状态"""
        self.get_logger().info(f'已处理{self.msg_count} 条消息')

def main(args=None):
    rclpy.init(args=args)
    node = SensorProcessor()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

运行验证：

# 终端1：运行节点
ros2 run my_nodes sensor_processor

# 终端2：发布测试数据
ros2 topic pub /raw_sensor_data std_msgs/msg/String '{data: "test_001"}' --once

# 终端3：查看处理结果
ros2 topic echo /processed_data
# 输出: data: '已处理: test_001'

三、生命周期节点（Managed Node）

普通节点启动就干活，停了就没了——没有中间状态。实际部署中，传感器需要先配置再激活，异常时需要安全停用，重启时需要清理再初始化。ROS2引入标准生命周期状态机解决这个问题。

生命周期状态机

四个主状态说明

状态	含义	允许操作
Unconfigured	节点已创建，未配置	只能执行 `configure` 转换
Inactive	已配置，未激活	可修改参数，不能收发消息
Active	正常运行	收发消息、执行回调
Finalized	已销毁	不可恢复，等待进程退出

关键约束：Inactive状态下发布者已创建但不生效，只有进入Active状态才真正收发消息。这保证了系统在所有节点就绪前不会产生不完整数据。

Python托管节点完整代码

import rclpy
from rclpy.node import Node
from rclpy.lifecycle import LifecycleNode, TransitionCallbackReturn
from rclpy.lifecycle.node import LifecycleState
from std_msgs.msg import String

class ManagedSensorNode(LifecycleNode):
    """生命周期托管传感器节点"""

    def __init__(self):
        super().__init__('managed_sensor')
        self.pub = None
        self.timer = None
        self.count = 0

    def on_configure(self, state: LifecycleState):
        """配置阶段：声明参数、创建发布者（不生效）"""
        self.get_logger().info('✅ on_configure: 正在配置...')
        self.declare_parameter('sensor_id', 'lidar_front')
        self.declare_parameter('publish_rate', 5.0)
        self.pub = self.create_lifecycle_publisher(String, 'sensor_output', 10)
        self.get_logger().info('✅ on_configure: 配置完成')
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: LifecycleState):
        """激活阶段：启动定时器，发布者生效"""
        self.get_logger().info('✅ on_activate: 正在激活...')
        rate = self.get_parameter('publish_rate').value
        self.timer = self.create_timer(1.0 / rate, self.publish_data)
        self.get_logger().info('✅ on_activate: 激活完成，开始发布')
        return super().on_activate(state)

    def on_deactivate(self, state: LifecycleState):
        """停用阶段：停止定时器，发布者暂停"""
        self.get_logger().info('✅ on_deactivate: 正在停用...')
        if self.timer:
            self.destroy_timer(self.timer)
            self.timer = None
        self.get_logger().info('✅ on_deactivate: 已停用')
        return super().on_deactivate(state)

    def on_cleanup(self, state: LifecycleState):
        """清理阶段：销毁发布者，回到未配置"""
        self.get_logger().info('✅ on_cleanup: 正在清理...')
        if self.pub:
            self.destroy_publisher(self.pub)
            self.pub = None
        self.count = 0
        self.get_logger().info('✅ on_cleanup: 清理完成')
        return TransitionCallbackReturn.SUCCESS

    def on_shutdown(self, state: LifecycleState):
        """关闭阶段：释放所有资源"""
        self.get_logger().info('✅ on_shutdown: 正在关闭...')
        if self.timer:
            self.destroy_timer(self.timer)
            self.timer = None
        if self.pub:
            self.destroy_publisher(self.pub)
            self.pub = None
        self.get_logger().info('✅ on_shutdown: 已关闭')
        return TransitionCallbackReturn.SUCCESS

    def publish_data(self):
        """定时发布传感器数据"""
        self.count += 1
        sensor_id = self.get_parameter('sensor_id').value
        msg = String()
        msg.data = f'[{sensor_id}] 数据 #{self.count}'
        self.pub.publish(msg)

def main(args=None):
    rclpy.init(args=args)
    node = ManagedSensorNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

命令行控制生命周期

# 启动托管节点（启动后处于 Unconfigured 状态）
ros2 run my_nodes managed_sensor

# 查看当前状态
ros2 lifecycle get /managed_sensor
# 输出: unconfigured [1]

# 手动触发状态转换
ros2 lifecycle set /managed_sensor configure   # Unconfigured → Inactive
ros2 lifecycle set /managed_sensor activate    # Inactive → Active
ros2 lifecycle set /managed_sensor deactivate  # Active → Inactive
ros2 lifecycle set /managed_sensor cleanup     # Inactive → Unconfigured

# 查看所有可用转换
ros2 lifecycle list /managed_sensor

注意：生命周期节点启动后不会自动发布消息，必须手动或通过Launch文件触发 configure 和 activate。这是初学者最常见的坑。

四、节点继承与组合

继承rclpy.node.Node

ROS2 Python节点的标准写法是继承rclpy.node.Node，在__init__中完成初始化：

import rclpy
from rclpy.node import Node

class MyCustomNode(Node):
    def __init__(self, node_name='my_node', **kwargs):
        # node_name可被外部覆盖，方便命名空间隔离
        super().__init__(node_name, **kwargs)
        # 在此初始化发布者、订阅者、定时器等

组合模式：一个进程运行多个节点

ROS2允许一个进程运行多个节点。两种方式：

手动spin多个节点：简单，但回调串行执行
Executor：支持单线程/多线程调度

import rclpy
from rclpy.node import Node
from rclpy.executors import SingleThreadedExecutor, MultiThreadedExecutor
from std_msgs.msg import String

class SensorA(Node):
    def __init__(self):
        super().__init__('sensor_a')
        self.pub = self.create_publisher(String, 'data_a', 10)
        self.timer = self.create_timer(0.5, self.publish)
        self.count = 0

    def publish(self):
        msg = String()
        self.count += 1
        msg.data = f'sensor_a: #{self.count}'
        self.pub.publish(msg)

class SensorB(Node):
    def __init__(self):
        super().__init__('sensor_b')
        self.pub = self.create_publisher(String, 'data_b', 10)
        self.timer = self.create_timer(0.8, self.publish)
        self.count = 0

    def publish(self):
        msg = String()
        self.count += 1
        msg.data = f'sensor_b: #{self.count}'
        self.pub.publish(msg)

def main(args=None):
    rclpy.init(args=args)

    node_a = SensorA()
    node_b = SensorB()

    # 方式1：单线程执行器（回调串行，简单安全）
    executor = SingleThreadedExecutor()
    executor.add_node(node_a)
    executor.add_node(node_b)

    # 方式2：多线程执行器（回调并行，适合耗时操作）
    # executor = MultiThreadedExecutor()
    # executor.add_node(node_a)
    # executor.add_node(node_b)

    try:
        executor.spin()
    finally:
        node_a.destroy_node()
        node_b.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Executor选择

Executor	回调方式	适用场景	线程安全
SingleThreadedExecutor	串行执行	节点间无竞争、回调快	天然安全
MultiThreadedExecutor	并行执行	回调耗时长、需实时响应	需加锁保护共享资源

选择原则：默认用SingleThreadedExecutor，只有当回调耗时长（如图像处理、网络请求）导致其他回调延迟时，才切换MultiThreadedExecutor，并注意加锁。
来自 linuxros.cn · linuxROS

五、节点命名与命名空间

命名规则

类型	格式	示例
节点名	`/<namespace>/<node_name>`	`/robot1/lidar_driver`
命名空间	`/`开头，`/`分隔层级	`/robot1/sensors`
全局名	以`/`开头的完整路径	`/robot1/sensors/lidar`

命名约束：只允许字母、数字、下划线，不能以数字开头，不能有连续下划线。

重映射（Remapping）

重映射允许在启动时修改话题名、服务名，不改代码就能适配不同系统配置。

# 基本重映射：将 /cmd_vel 映射到 /robot1/cmd_vel
ros2 run my_nodes controller --ros-args \
  -r cmd_vel:=/robot1/cmd_vel

# 命名空间：给节点加命名空间前缀
ros2 run my_nodes controller --ros-args \
  --remap __ns:=/robot1

# 修改节点名
ros2 run my_nodes controller --ros-args \
  --remap __node:=left_controller

# 组合使用：命名空间 + 重映射
ros2 run my_nodes controller --ros-args \
  -r __ns:=/robot1 \
  -r cmd_vel:=/robot1/cmd_vel

多机器人命名空间隔离

# 机器人1的控制器
ros2 run my_pkg controller --ros-args \
  -r __ns:=/robot1 \
  -r __node:=controller

# 机器人2的控制器
ros2 run my_pkg controller --ros-args \
  -r __ns:=/robot2 \
  -r __node:=controller

# 查看节点列表
ros2 node list
# /robot1/controller
# /robot2/controller

同一个可执行文件，通过命名空间和节点名重映射，运行两个独立实例，话题自动隔离。

六、常见问题

Q1：生命周期节点启动后不发布消息？

生命周期节点必须经过 configure 和 activate 才能工作。启动后它处于 Unconfigured 状态，什么都不做。解决方案：

# 手动激活
ros2 lifecycle set /managed_sensor configure
ros2 lifecycle set /managed_sensor activate

或在Launch文件中使用 LifecycleNode 自动管理转换。

Q2：多节点单进程时回调阻塞？

SingleThreadedExecutor串行执行回调，一个回调耗时长会阻塞其他回调。解决方案：切换MultiThreadedExecutor，并对共享资源加锁：

import threading
from rclpy.executors import MultiThreadedExecutor

class ThreadSafeNode(Node):
    def __init__(self):
        super().__init__('thread_safe_node')
        self.lock = threading.Lock()
        self.shared_data = {}

    def callback_a(self, msg):
        with self.lock:
            self.shared_data['key'] = msg.data

    def callback_b(self, msg):
        with self.lock:
            value = self.shared_data.get('key', 'default')

Q3：节点名冲突？

同一进程中创建两个同名节点会报错。不同进程的同名节点会互相覆盖注册信息。解决方案：用命名空间隔离：

ros2 run my_pkg sensor --ros-args -r __ns:=/front
ros2 run my_pkg sensor --ros-args -r __ns:=/rear
# 结果: /front/sensor 和 /rear/sensor，互不冲突

七、总结

节点是ROS2的积木块，生命周期是让积木块行为可控的规则。普通节点适合简单场景，生命周期节点适合需要安全启停的工业部署。多节点组合时，Executor决定回调调度策略，命名空间解决多实例冲突。

开发场景	推荐模式
快速原型、简单功能	普通Node + SingleThreadedExecutor
传感器驱动、需安全启停	LifecycleNode + Launch自动管理
多节点协作、回调耗时	多节点 + MultiThreadedExecutor
多机器人/多实例	命名空间隔离 + 重映射
生产部署	LifecycleNode + 参数约束 + 错误处理

本文首发于linuxros.cn，转载请注明出处。