ROS2系列教程：认识节点与ROS图

1
# minimal_node.py —— 最简 rclpy 节点
2
import rclpy
3
from rclpy.node import Node
4

5
class MyNode(Node):
6
    def __init__(self):
7
        # 第一个参数是节点名称，第二个参数是命名空间（可省略）
8
        super().__init__('my_python_node')
9
        self.get_logger().info('Python 节点已启动！')
10

11
def main(args=None):
12
    rclpy.init(args=args)          # 1. 初始化 rclpy
13
    node = MyNode()                # 2. 创建节点实例
14
    rclpy.spin(node)               # 3. 进入事件循环（等待回调）
15
    node.destroy_node()            # 4. 清理节点
16
    rclpy.shutdown()               # 5. 关闭 rclpy
17

18
if __name__ == '__main__':
19
    main()

1
[INFO] [1690000000.123456789] [my_python_node]: Python 节点已启动！

1
# namespace_node.py —— 演示命名空间
2
import rclpy
3
from rclpy.node import Node
4

5
def main():
6
    rclpy.init()
7
    # 节点全名 = 命名空间 + 节点名，即 /robot_arm/controller
8
    node = Node('controller', namespace='/robot_arm')
9
    node.get_logger().info(f'节点全名: {node.get_name()}')
10
    node.get_logger().info(f'命名空间: {node.get_namespace()}')
11
    node.get_logger().info(f'带命名空间的全名: {node.get_fully_qualified_name()}')
12
    rclpy.spin(node)
13
    node.destroy_node()
14
    rclpy.shutdown()
15

16
if __name__ == '__main__':
17
    main()

1
# timer_node.py —— 定时器节点
2
import rclpy
3
from rclpy.node import Node
4

5
class TimerNode(Node):
6
    def __init__(self):
7
        super().__init__('timer_node')
8
        self.count = 0
9
        # 每 1.0 秒触发一次回调（注意单位是秒，浮点数）
10
        self.timer = self.create_timer(1.0, self.timer_callback)
11

12
    def timer_callback(self):
13
        self.count += 1
14
        self.get_logger().info(f'第 {self.count} 次心跳')
15

16
def main():
17
    rclpy.init()
18
    node = TimerNode()
19
    rclpy.spin(node)
20
    node.destroy_node()
21
    rclpy.shutdown()
22

23
if __name__ == '__main__':
24
    main()

1
// minimal_node.cpp —— 最简 rclcpp 节点
2
#include "rclcpp/rclcpp.hpp"
3

4
class MyNode : public rclcpp::Node
5
{
6
public:
7
  MyNode() : Node("my_cpp_node")
8
  {
9
    RCLCPP_INFO(this->get_logger(), "C++ 节点已启动！");
10
  }
11
};
12

13
int main(int argc, char** argv)
14
{
15
  rclcpp::init(argc, argv);              // 1. 初始化
16
  auto node = std::make_shared<MyNode>(); // 2. 创建节点（智能指针）
17
  rclcpp::spin(node);                    // 3. 事件循环
18
  rclcpp::shutdown();                    // 4. 关闭
19
  return 0;
20
}

1
// timer_node.cpp —— C++ 定时器节点
2
#include "rclcpp/rclcpp.hpp"
3
#include <chrono>
4

5
class TimerNode : public rclcpp::Node
6
{
7
public:
8
  TimerNode() : Node("timer_node_cpp"), count_(0)
9
  {
10
    // 1.0 秒周期，单位是 std::chrono::duration
11
    timer_ = this->create_wall_timer(
12
      std::chrono::seconds(1),
13
      std::bind(&TimerNode::timer_callback, this));
14
  }
15

16
private:
17
  void timer_callback()
18
  {
19
    count_++;
20
    RCLCPP_INFO(this->get_logger(), "第 %ld 次心跳", count_);
21
  }
22

23
  rclcpp::TimerBase::SharedPtr timer_;
24
  long count_;
25
};
26

27
int main(int argc, char** argv)
28
{
29
  rclcpp::init(argc, argv);
30
  rclcpp::spin(std::make_shared<TimerNode>());
31
  rclcpp::shutdown();
32
  return 0;
33
}

1
初始化(init) → 构造(创建Node实例) → 配置(声明参数/创建端点) → 运行(spin事件循环) → 销毁(destroy/shutdown)

1
while (节点存活) {
2
    从回调队列取出一个待执行的回调;   // 可能来自话题消息、服务请求、定时器触发
3
    执行该回调;                        // 默认单线程，一次只执行一个
4
}

1
# multi_thread.py —— 多线程执行器
2
import rclpy
3
import threading
4
from rclpy.node import Node
5
from rclpy.executors import MultiThreadedExecutor
6

7
class BusyNode(Node):
8
    def __init__(self):
9
        super().__init__('busy_node')
10
        self.timer_a = self.create_timer(0.5, self.cb_a)
11
        self.timer_b = self.create_timer(0.5, self.cb_b)
12

13
    def cb_a(self):
14
        # 模拟耗时 3 秒的阻塞操作
15
        import time; time.sleep(3)
16
        self.get_logger().info('A 完成')
17

18
    def cb_b(self):
19
        self.get_logger().info('B 执行')
20

21
def main():
22
    rclpy.init()
23
    node = BusyNode()
24
    executor = MultiThreadedExecutor(num_threads=4)  # 4 线程
25
    executor.add_node(node)
26
    # 在子线程中运行执行器
27
    t = threading.Thread(target=executor.spin, daemon=True)
28
    t.start()
29
    try:
30
        import time
31
        while True:
32
            time.sleep(1)
33
    except KeyboardInterrupt:
34
        executor.shutdown()
35
    rclpy.shutdown()
36

37
if __name__ == '__main__':
38
    main()

1
# param_node.py —— 参数演示（Python）
2
import rclpy
3
from rclpy.node import Node
4
from rclpy.parameter import Parameter
5

6
class ParamNode(Node):
7
    def __init__(self):
8
        super().__init__('param_node')
9
        # 声明参数：名称 + 默认值 + 描述
10
        self.declare_parameter('robot_name', 'turtle_01')
11
        self.declare_parameter('max_speed', 2.5)
12
        self.declare_parameter('verbose', True)
13
        # 读取参数（返回 Parameter 对象，用 .value 取值）
14
        self.robot_name = self.get_parameter('robot_name').value
15
        self.max_speed = self.get_parameter('max_speed').value
16
        self.get_logger().info(
17
            f'机器人: {self.robot_name}, 最大速度: {self.max_speed}')
18

19
def main():
20
    rclpy.init()
21
    node = ParamNode()
22
    rclpy.spin(node)
23
    node.destroy_node()
24
    rclpy.shutdown()
25

26
if __name__ == '__main__':
27
    main()

1
// param_node.cpp —— 参数演示（C++）
2
#include "rclcpp/rclcpp.hpp"
3

4
class ParamNode : public rclcpp::Node
5
{
6
public:
7
  ParamNode() : Node("param_node_cpp")
8
  {
9
    this->declare_parameter("robot_name", "turtle_01");
10
    this->declare_parameter("max_speed", 2.5);
11
    auto name = this->get_parameter("robot_name").as_string();
12
    auto speed = this->get_parameter("max_speed").as_double();
13
    RCLCPP_INFO(this->get_logger(), "机器人: %s, 最大速度: %.1f",
14
                name.c_str(), speed);
15
  }
16
};
17

18
int main(int argc, char** argv)
19
{
20
  rclcpp::init(argc, argv);
21
  rclcpp::spin(std::make_shared<ParamNode>());
22
  rclcpp::shutdown();
23
  return 0;
24
}

1
# 在节点内输出各级日志
2
node.get_logger().debug('调试信息（默认不显示）')
3
node.get_logger().info('普通信息')
4
node.get_logger().warn('警告信息')
5
node.get_logger().error('错误信息')
6
node.get_logger().fatal('致命错误')

1
multi_node_demo/
2
├── heartbeat.py      # 心跳节点（定时器 + 参数）
3
├── monitor.py        # 监视节点（定时器读取参数）
4
└── start.launch.py   # launch 文件同时启动两个节点

1
# heartbeat.py —— 心跳节点
2
import rclpy
3
from rclpy.node import Node
4

5
class HeartbeatNode(Node):
6
    def __init__(self):
7
        super().__init__('heartbeat')
8
        self.declare_parameter('rate', 1.0)      # 心跳频率参数
9
        self.declare_parameter('heartbeat', 0)   # 心跳计数（共享用）
10
        self.count = 0
11
        rate = self.get_parameter('rate').value
12
        self.timer = self.create_timer(rate, self.beat)
13

14
    def beat(self):
15
        self.count += 1
16
        # 把计数写回参数（同进程内其他节点可读，简单演示用）
17
        self.set_parameters([rclpy.parameter.Parameter(
18
            'heartbeat', rclpy.parameter.Parameter.Type.INTEGER, self.count)])
19
        self.get_logger().info(f'心跳 #{self.count}')
20

21
def main():
22
    rclpy.init()
23
    node = HeartbeatNode()
24
    rclpy.spin(node)
25
    node.destroy_node()
26
    rclpy.shutdown()
27

28
if __name__ == '__main__':
29
    main()

1
# monitor.py —— 监视节点
2
import rclpy
3
from rclpy.node import Node
4

5
class MonitorNode(Node):
6
    def __init__(self):
7
        super().__init__('monitor')
8
        self.declare_parameter('heartbeat', 0)
9
        # 每 3 秒检查一次心跳值
10
        self.timer = self.create_timer(3.0, self.check)
11

12
    def check(self):
13
        hb = self.get_parameter('heartbeat').value
14
        self.get_logger().info(f'当前心跳计数: {hb}')
15

16
def main():
17
    rclpy.init()
18
    node = MonitorNode()
19
    rclpy.spin(node)
20
    node.destroy_node()
21
    rclpy.shutdown()
22

23
if __name__ == '__main__':
24
    main()

1
# start.launch.py —— 同时启动心跳与监视节点
2
from launch import LaunchDescription
3
from launch_ros.actions import Node
4

5
def generate_launch_description():
6
    return LaunchDescription([
7
        Node(package='demo_nodes_py', executable='heartbeat',
8
             name='heartbeat'),
9
        Node(package='demo_nodes_py', executable='monitor',
10
             name='monitor'),
11
    ])