xs
6.8 KiB
6.8 KiB
17 · 生命周期节点 (Lifecycle Node) 完全指南
目标: 理解 ROS2 Lifecycle Node 的设计、状态机、转换机制,能用 Lifecycle Node 管理资源密集型节点。
目录
1. 是什么
Lifecycle Node 是 rclcpp / rclpy 的特殊节点基类,提供:
- 受控状态切换(configure → activate → deactivate → cleanup → shutdown)
- 每种状态都有自己的回调(on_configure / on_activate 等)
- 外部触发转换(通过 service 调用)
适用场景:
- 资源密集型节点(加载 ML 模型、连接硬件、订阅话题)
- 需要明确"启动顺序"的复杂系统(相机先 ready,处理节点再 activate)
- 安全敏感系统(切换前确认硬件 OK)
2. 为什么需要
普通节点的缺点:
# ❌ 普通节点 — 启动即订阅 / 订阅即消费 / 死了就完了
class MyNode(Node):
def __init__(self):
super().__init__('my_node')
self._sub = self.create_subscription(...) # 一启动就开始消费
问题:
- 想"暂停"接收?做不到
- 想"重新初始化"?得 kill 重启
- 想"加载模型失败就停"?只能异常退出
Lifecycle Node 解决:
# ✅ Lifecycle Node — 显式状态切换
class MyLifecycleNode(LifecycleNode):
def on_configure(self, state):
# 加载模型、分配资源
# 失败 → return FAILURE,不会进入 active
def on_activate(self, state):
# 订阅话题、启动定时器
# 失败 → return FAILURE,自动 cleanup
def on_deactivate(self, state):
# 停止订阅、暂停定时器(但资源仍在)
def on_cleanup(self, state):
# 释放模型、断开连接(回到 unconfigured)
3. 状态机
configure
unconfigured ───────→ inactive
▲ │ │ activate
│ │ cleanup ▼
│ └────────────── active
│ │ deactivate
└──────────────────────┘
shutdown(任何状态都可触发)→ finalized
四个主要状态:
unconfigured: 已创建但未配置inactive: 已配置但未激活active: 完全运行(处理数据)finalized: 终止(不可逆)
转换事件(transition):
configure: unconfigured → inactiveactivate: inactive → activedeactivate: active → inactivecleanup: inactive → unconfiguredshutdown: 任何 → finalized
4. 转换回调
每个转换回调签名: (state: State) -> TransitionCallbackReturn
返回:
SUCCESS: 转换成功,进入目标状态FAILURE: 转换失败,回到原状态ERROR: 转换错误,直接进 finalized
必须重写:
on_configure(state)on_activate(state)on_deactivate(state)on_cleanup(state)on_shutdown(state)
可选重写:
on_error(state): 错误处理
5. Service 接口
每个 Lifecycle Node 自动注册两个 service:
/<node>/change_state(lifecycle_msgs/srv/ChangeState)— 触发转换/<node>/get_state(lifecycle_msgs/srv/GetState)— 查询状态
# 查状态
ros2 service call /lifecycle_demo_node/get_state lifecycle_msgs/srv/GetState
# 触发 configure (transition id = 1)
ros2 service call /lifecycle_demo_node/change_state \
lifecycle_msgs/srv/ChangeState "{transition: {id: 1}}"
Transition IDs:
| ID | 转换 |
|---|---|
| 0 | configure |
| 1 | cleanup |
| 2 | activate |
| 3 | deactivate |
| 4 | shutdown |
6. Python 实现
from rclpy.lifecycle import LifecycleNode, TransitionCallbackReturn
class LifecycleDemoNode(LifecycleNode):
def on_configure(self, state):
self._publisher = self.create_lifecycle_publisher(String, 'topic', 10)
return TransitionCallbackReturn.SUCCESS
def on_activate(self, state):
self._timer = self.create_timer(0.5, self._publish)
return super().on_activate(state)
def on_deactivate(self, state):
self.destroy_timer(self._timer)
return super().on_deactivate(state)
def on_cleanup(self, state):
self.destroy_lifecycle_publisher(self._publisher)
return TransitionCallbackReturn.SUCCESS
7. C++ 实现
#include "rclcpp_lifecycle/lifecycle_node.hpp"
class LifecycleDemoNode : public rclcpp_lifecycle::LifecycleNode
{
public:
LifecycleDemoNode() : rclcpp_lifecycle::LifecycleNode("demo") {}
rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn
on_configure(const rclcpp_lifecycle::State &)
{
publisher_ = this->create_publisher<String>("topic", 10);
return rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS;
}
};
8. CLI 控制
# 1. 启 Lifecycle Node
ros2 launch my_pkg lifecycle_demo.py
# 2. 触发 configure
ros2 lifecycle set /lifecycle_node configure
# (Humble 后 ros2 lifecycle set 直接用,而不是 service call)
# 3. 触发 activate
ros2 lifecycle set /lifecycle_node activate
# 4. 看状态
ros2 lifecycle get /lifecycle_node
# 输出:active [3]
9. 实战模式
模式 1: ML 模型加载
def on_configure(self, state):
try:
self._model = torch.load('model.pt') # 加载 ML 模型
return TransitionCallbackReturn.SUCCESS
except FileNotFoundError:
return TransitionCallbackReturn.FAILURE # 配置失败,节点不可用
def on_activate(self, state):
self._sub = self.create_subscription(Image, 'image_raw', self._infer, 10)
return super().on_activate(state)
def _infer(self, msg):
if self._model is None: return # 不会到这里(因为 activate 前要 configure 成功)
result = self._model(msg)
...
模式 2: 硬件连接(相机)
def on_configure(self, state):
try:
self._camera = cv2.VideoCapture(0)
if not self._camera.isOpened():
return TransitionCallbackReturn.FAILURE
return TransitionCallbackReturn.SUCCESS
except Exception:
return TransitionCallbackReturn.FAILURE
def on_cleanup(self, state):
if self._camera:
self._camera.release()
return TransitionCallbackReturn.SUCCESS