🧩 第02课：节点(Node)

📋 课程目标

• 深入理解ROS2节点的概念与作用
• 掌握Python和C++节点的编写方法
• 理解生命周期节点(Lifecycle Node)的原理
• 学会使用ros2命令行工具管理节点
• 掌握节点间通信的基本模式

🧠 什么是节点？

节点(Node)是ROS2中最基本的执行单元。每个节点都是一个独立的进程，负责完成一个单一的功能模块。节点之间通过话题、服务、动作等方式进行通信，共同构成复杂的机器人系统。

📐 节点架构图

🐍 Python节点开发

基础节点模板

#!/usr/bin/env python3
"""基础ROS2节点模板"""

import rclpy
from rclpy.node import Node


class MinimalNode(Node):
    """最简ROS2节点"""

    def __init__(self):
        # 必须调用父类构造器，指定节点名
        super().__init__('minimal_node')
        self.get_logger().info('节点已启动！')


def main(args=None):
    rclpy.init(args=args)          # 1. 初始化rclpy
    node = MinimalNode()            # 2. 创建节点
    rclpy.spin(node)                # 3. 保持运行，处理回调
    node.destroy_node()             # 4. 销毁节点
    rclpy.shutdown()                # 5. 关闭rclpy


if __name__ == '__main__':
    main()

传感器模拟节点（含发布者+定时器+参数）

#!/usr/bin/env python3
"""温度传感器模拟节点 - 演示节点核心功能"""

import math
import random
import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
from sensor_msgs.msg import Temperature
from std_msgs.msg import Header
from builtin_interfaces.msg import Time


class TemperatureSensorNode(Node):
    """模拟温度传感器节点"""

    def __init__(self):
        super().__init__('temperature_sensor')

        # 声明参数（可在启动时覆盖）
        self.declare_parameter('base_temp', 25.0)       # 基准温度
        self.declare_parameter('noise_amplitude', 2.0)   # 噪声幅度
        self.declare_parameter('publish_rate', 1.0)      # 发布频率Hz

        # 获取参数值
        base_temp = self.get_parameter('base_temp').value
        self.noise_amp = self.get_parameter('noise_amplitude').value
        rate = self.get_parameter('publish_rate').value

        # QoS配置：传感器数据适合BEST_EFFORT
        qos_profile = QoSProfile(
            reliability=ReliabilityPolicy.BEST_EFFORT,
            history=HistoryPolicy.KEEP_LAST,
            depth=5
        )

        # 创建发布者
        self.temp_pub = self.create_publisher(
            Temperature, 'temperature', qos_profile
        )

        # 创建定时器
        period = 1.0 / rate
        self.timer = self.create_timer(period, self.publish_temperature)

        # 状态变量
        self.time_elapsed = 0.0
        self.base_temp = base_temp

        self.get_logger().info(
            f'温度传感器节点启动 - 基准温度: {base_temp}°C, '
            f'频率: {rate}Hz'
        )

    def publish_temperature(self):
        """定时发布温度数据"""
        # 模拟温度变化（正弦波 + 随机噪声）
        self.time_elapsed += 0.1
        sin_component = math.sin(self.time_elapsed) * 3.0
        noise = random.uniform(-self.noise_amp, self.noise_amp)
        temp = self.base_temp + sin_component + noise

        # 构建消息
        msg = Temperature()
        msg.header = Header()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.header.frame_id = 'sensor_frame'
        msg.temperature = temp
        msg.variance = self.noise_amp ** 2

        self.temp_pub.publish(msg)
        self.get_logger().debug(f'发布温度: {temp:.2f}°C')


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


if __name__ == '__main__':
    main()

节点组合器（含订阅者+发布者）

#!/usr/bin/env python3
"""温度监控节点 - 订阅温度数据，发布告警"""

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Temperature
from std_msgs.msg import String, Bool


class TemperatureMonitorNode(Node):
    """温度监控节点：订阅温度 → 判断阈值 → 发布告警"""

    def __init__(self):
        super().__init__('temperature_monitor')

        # 参数
        self.declare_parameter('warning_threshold', 30.0)
        self.declare_parameter('critical_threshold', 35.0)

        self.warning_threshold = self.get_parameter('warning_threshold').value
        self.critical_threshold = self.get_parameter('critical_threshold').value

        # 订阅温度话题
        self.temp_sub = self.create_subscription(
            Temperature, 'temperature', self.temp_callback, 10
        )

        # 发布告警话题
        self.alert_pub = self.create_publisher(String, 'temperature_alert', 10)
        self.fan_ctrl_pub = self.create_publisher(Bool, 'fan_control', 10)

        self.get_logger().info(
            f'温度监控启动 - 告警阈值: {self.warning_threshold}°C, '
            f'危险阈值: {self.critical_threshold}°C'
        )

    def temp_callback(self, msg):
        """温度数据回调"""
        temp = msg.temperature

        alert_msg = String()
        fan_msg = Bool()

        if temp >= self.critical_threshold:
            alert_msg.data = f'🔴 危险！温度 {temp:.1f}°C 超过临界值！'
            fan_msg.data = True
            self.get_logger().error(alert_msg.data)
        elif temp >= self.warning_threshold:
            alert_msg.data = f'🟡 警告：温度 {temp:.1f}°C 偏高'
            fan_msg.data = True
            self.get_logger().warn(alert_msg.data)
        else:
            alert_msg.data = f'🟢 正常：温度 {temp:.1f}°C'
            fan_msg.data = False
            self.get_logger().info(alert_msg.data)

        self.alert_pub.publish(alert_msg)
        self.fan_ctrl_pub.publish(fan_msg)


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


if __name__ == '__main__':
    main()

🔧 C++节点开发

// temperature_sensor_cpp.cpp - C++版本温度传感器节点
#include "rclcpp/rclcpp.hpp"
#include "sensor_msgs/msg/temperature.hpp"
#include "std_msgs/msg/header.hpp"

#include <cmath>
#include <random>
#include <chrono>

using namespace std::chrono_literals;

class TemperatureSensorCpp : public rclcpp::Node {
public:
    TemperatureSensorCpp() : Node("temperature_sensor_cpp"), time_elapsed_(0.0) {
        // 声明参数
        this->declare_parameter("base_temp", 25.0);
        this->declare_parameter("noise_amplitude", 2.0);
        this->declare_parameter("publish_rate", 1.0);

        double base_temp = this->get_parameter("base_temp").as_double();
        noise_amp_ = this->get_parameter("noise_amplitude").as_double();
        double rate = this->get_parameter("publish_rate").as_double();
        base_temp_ = base_temp;

        // 创建发布者
        temp_pub_ = this->create_publisher<sensor_msgs::msg::Temperature>(
            "temperature_cpp", 10);

        // 创建定时器
        auto period = std::chrono::duration<double>(1.0 / rate);
        timer_ = this->create_wall_timer(
            std::chrono::duration_cast<std::chrono::nanoseconds>(period),
            std::bind(&TemperatureSensorCpp::publish_temperature, this));

        RCLCPP_INFO(this->get_logger(),
            "C++温度传感器启动 - 基准温度: %.1f°C, 频率: %.1fHz",
            base_temp, rate);
    }

private:
    void publish_temperature() {
        time_elapsed_ += 0.1;
        double sin_comp = std::sin(time_elapsed_) * 3.0;
        double noise = noise_dist_(rng_);
        double temp = base_temp_ + sin_comp + noise;

        auto msg = sensor_msgs::msg::Temperature();
        msg.header.stamp = this->now();
        msg.header.frame_id = "sensor_frame_cpp";
        msg.temperature = temp;
        msg.variance = noise_amp_ * noise_amp_;

        temp_pub_->publish(msg);
        RCLCPP_DEBUG(this->get_logger(), "发布温度: %.2f°C", temp);
    }

    rclcpp::Publisher<sensor_msgs::msg::Temperature>::SharedPtr temp_pub_;
    rclcpp::TimerBase::SharedPtr timer_;
    double base_temp_;
    double noise_amp_;
    double time_elapsed_;
    std::mt19937 rng_{std::random_device{}()};
    std::uniform_real_distribution<double> noise_dist_{-noise_amp_, noise_amp_};
};

int main(int argc, char *argv[]) {
    rclcpp::init(argc, argv);
    rclcpp::spin(std::make_shared<TemperatureSensorCpp>());
    rclcpp::shutdown();
    return 0;
}

🔄 生命周期节点(Lifecycle Node)

生命周期节点是ROS2引入的重要特性，它定义了节点的状态机，使节点管理更加可控。

Python生命周期节点实现

#!/usr/bin/env python3
"""生命周期节点示例 - 电机控制器"""

import rclpy
from rclpy.lifecycle import LifecycleNode, LifecycleState, TransitionCallbackReturn
from std_msgs.msg import Float64


class MotorControllerLifecycleNode(LifecycleNode):
    """电机控制器生命周期节点"""

    def __init__(self):
        super().__init__('motor_controller_lifecycle')

        # 在Unconfigured状态，不能创建发布者/订阅者
        self.motor_pub = None
        self.speed = 0.0
        self.is_running = False

    def on_configure(self, state: LifecycleState) -> TransitionCallbackReturn:
        """配置阶段：创建发布者、订阅者，但还不发布数据"""
        self.get_logger().info('🔧 配置电机控制器...')
        self.motor_pub = self.create_lifecycle_publisher(Float64, 'motor_speed', 10)
        self.speed = 0.0

        # 模拟硬件初始化
        self.get_logger().info('  硬件初始化完成')
        self.get_logger().info('  通信检查通过')
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: LifecycleState) -> TransitionCallbackReturn:
        """激活阶段：开始发布数据"""
        self.get_logger().info('✅ 激活电机控制器...')
        self.is_running = True

        # 启动控制循环
        self.timer = self.create_timer(0.1, self.control_loop)

        return super().on_activate(state)

    def on_deactivate(self, state: LifecycleState) -> TransitionCallbackReturn:
        """停用阶段：停止发布，但保留资源"""
        self.get_logger().info('⏸️ 停用电机控制器...')
        self.is_running = False
        self.destroy_timer(self.timer)

        return super().on_deactivate(state)

    def on_cleanup(self, state: LifecycleState) -> TransitionCallbackReturn:
        """清理阶段：释放所有资源"""
        self.get_logger().info('🧹 清理电机控制器...')
        self.destroy_publisher(self.motor_pub)
        self.motor_pub = None
        return TransitionCallbackReturn.SUCCESS

    def on_shutdown(self, state: LifecycleState) -> TransitionCallbackReturn:
        """关闭阶段"""
        self.get_logger().info('👋 关闭电机控制器...')
        if self.motor_pub is not None:
            self.destroy_publisher(self.motor_pub)
        return TransitionCallbackReturn.SUCCESS

    def control_loop(self):
        """控制循环"""
        if not self.is_running:
            return

        msg = Float64()
        msg.data = self.speed
        self.motor_pub.publish(msg)
        self.get_logger().debug(f'电机速度: {self.speed}')


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


if __name__ == '__main__':
    main()

管理生命周期节点的命令：

# 查看生命周期节点状态
ros2 lifecycle get /motor_controller_lifecycle

# 触发状态转换
ros2 lifecycle set /motor_controller_lifecycle configure
ros2 lifecycle set /motor_controller_lifecycle activate
ros2 lifecycle set /motor_controller_lifecycle deactivate
ros2 lifecycle set /motor_controller_lifecycle cleanup

# 查看可用转换
ros2 lifecycle list /motor_controller_lifecycle

🖥️ ros2节点命令行工具

🔗 节点命名规范

# ✅ 合法命名
temperature_sensor
motor_controller
lidar_driver

# ❌ 非法命名
_TempSensor      # 不能以下划线开头
temp-sensor      # 不能包含连字符
Temp__Sensor     # 不能有连续下划线
tempSensor_      # 不能以下划线结尾

命名空间

命名空间用于组织节点，避免命名冲突：

# 启动时指定命名空间
ros2 run my_pkg temperature_sensor --ros-args \
  --remap __ns:=/robot1/sensors

# 命名空间中的节点
# /robot1/sensors/temperature_sensor
# /robot2/sensors/temperature_sensor

🔄 节点组合(Component Node)

ROS2支持将多个节点在同一个进程中运行，减少通信开销：

#!/usr/bin/env python3
"""组件化节点 - 多节点在同一进程"""

import rclpy
from rclpy.executors import SingleThreadedExecutor, MultiThreadedExecutor
import threading


class SensorNode(rclpy.node.Node):
    def __init__(self):
        super().__init__('sensor_node')
        self.pub = self.create_publisher(std_msgs.msg.String, 'sensor_data', 10)
        self.timer = self.create_timer(0.5, self.publish_data)

    def publish_data(self):
        msg = std_msgs.msg.String()
        msg.data = 'sensor reading'
        self.pub.publish(msg)


class ProcessorNode(rclpy.node.Node):
    def __init__(self):
        super().__init__('processor_node')
        self.sub = self.create_subscription(
            std_msgs.msg.String, 'sensor_data', self.callback, 10)

    def callback(self, msg):
        self.get_logger().info(f'处理: {msg.data}')


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

    # 创建多个节点
    sensor = SensorNode()
    processor = ProcessorNode()

    # 使用多线程执行器同时运行两个节点
    executor = MultiThreadedExecutor()
    executor.add_node(sensor)
    executor.add_node(processor)

    try:
        executor.spin()
    finally:
        sensor.destroy_node()
        processor.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

📊 节点通信模型对比

🎯 练习题

# 终端1
ros2 run my_pkg temperature_sensor --ros-args --remap __ns:=/room1

# 终端2
ros2 run my_pkg temperature_sensor --ros-args --remap __ns:=/room2

# 查看
ros2 node list

🏆 成就解锁