节点：ROS2 里的"一个岗位"

01问题引入：一台机器人里，到底有多少个程序在干活

先别看代码，看一个更实在的问题。你让小车"到桌前把那瓶水拿过来"，它身上同时发生着这些事：

这五件事，没有一件能等另一件慢慢做完——相机不会为了等规划算完就停产 30 秒。它们必须同时跑、各干各的、只把结果交出去。数下来，一台能干活的机器人身上通常同时跑着 20 个以上这样的程序。

但这个类比有三处会带你走偏，先划清边界

类比只是脚手架，不能当定义用。上面这套"岗位"说法帮你快速理解"为什么要拆"，但它有三条不成立的边界——每一条后面都会展开，这里先记住正确说法：

# 终端 A：发布者
source /opt/ros/humble/setup.bash
ros2 run demo_nodes_cpp talker

# 终端 B：订阅者
source /opt/ros/humble/setup.bash
ros2 run demo_nodes_cpp listener

# -y 表示自动确认，省掉一次 y/n 交互
sudo apt install -y ros-humble-rqt-graph

0.3把计算图画出来终端 C

rqt_graph 是 ROS2 的"看图窗口"，完整用法 EXP-14 细讲，这里只要会打开、会读箭头。

source /opt/ros/humble/setup.bash
ros2 run rqt_graph rqt_graph

✔ 三个观察点逐个对：① 图里有两个椭圆 /talker、/listener；② 中间是矩形 /chatter，左边一支箭头从 talker 指进来，右边一支从它指去 listener（发布 → 话题 → 订阅）；③ 回终端 A 按 Ctrl+C，talker 的椭圆和箭头立刻从图上消失，重新 run 又自己连回来。

那这 20 多个节点，是随便乱拆的吗

不是。真实机器人里的节点有一条约定的分工纪律，按"信号往哪个方向走"分成四层。上层只做判断、不碰硬件；下层只碰硬件、不做判断——记住这一句，下面这张图就全看懂了。

信号在层间单向流动：感知 → 定位 → 决策 → 执行。真正让轮子转起来的那条链是 lidar_node → amcl_node → nav2_planner → nav2_controller → base_driver；而 mission_node 只发目标、不下命令。四层架构是通行惯例而非 ROS2 的强制规定——但正因为它，任何一层换实现时，其他层一行代码都不用改。

那为什么非要拆成这么多程序？

先别急着接受"要拆"，我们看一个真会发生的场景——如果硬把这三件事写进同一个程序，会怎样：

图 1-2｜拆与不拆，差别不在"代码好不好看"，而在于故障会不会扩散、节奏会不会互相拖累。这也解释了为什么机器人软件几乎一定得多进程。

所以 ROS2 的解法就是"拆岗位"：拆成多个独立进程，各自独立崩溃域、独立执行节奏、可选各自的语言，再用标准接口对话。代价也很实在——你要多管一份进程、多一份通信开销，所以不是"越碎越好"，而是按职责边界拆（怎么拆见下节的四层）。

但"拆成独立进程"不等于"一个进程塞一个节点"

这两件事经常被混为一谈，先把它们分清楚：

但 ROS2 并不要求这样。同一个程序里创建多个节点完全可以，还能省掉进程间传输开销。看这段 rclpy 示例：

import rclpy
from rclpy.node import Node

# ── 同一个 context（同一个进程、同一套 ROS 运行环境）──
ctx1 = rclpy.Context()
rclpy.init(context=ctx1)
node_a = Node('a', context=ctx1)   # 节点 a
node_b = Node('b', context=ctx1)   # 节点 b —— 与 a 同进程、同 context
# 此时 `ros2 node list` 看到 2 个节点，但系统里只多了 1 个进程

# ── 再起一个 context：仍在这一个进程里，但是彼此独立的 ROS 实例 ──
ctx2 = rclpy.Context()
rclpy.init(context=ctx2)
node_c = Node('c', context=ctx2)   # 节点 c：同进程，但属于另一套 ROS 运行环境

# 结果：3 个节点全部在同一个 Python 进程内 —— 进程 1 个，节点 3 个

02核心原理：跟着一个真实节点，把生命周期走一遍

上一节说"节点是岗位"。可岗位怎么设立、怎么开始干活、怎么下班？我们直接拿一个真节点当标本——gas_sensor_node，实验箱上读 MQ5 气体传感器、每秒报一次浓度的那个程序。

全部代码不到 30 行，去掉 import 只剩四块。先扫一眼，再一块块拆：

from rclpy.node import Node

class GasSensorNode(Node):                 # 块1：继承 Node，拿到"入场券"
    def __init__(self):
        super().__init__('gas_sensor_node')      # ① 报名字 → 进 ROS2 图
        self.declare_parameter('publish_period', 1.0)  # ② 声明旋钮
        self.pub = self.create_publisher(Float32, '/gas_sensor', qos)  # ③ 开出口
        self.timer = self.create_timer(period, self.timer_cb)          # ④ 定闹钟
        # 注：period 在 __init__ 里就能读到，因为 super().__init__() 已执行完，
        #     节点此时已完整；但声明(②)一定要在读(period)之前。

    def timer_cb(self):                       # 块3：回调，被闹钟触发
        raw, v, ppm, err = read_mq5_once()
        self.pub.publish(Float32(data=ppm))
        self.get_logger().info(f'raw={raw} voltage={v:.3f}V ppm={ppm:.2f}')

def main(args=None):                         # 块4：入口，固定四步
    rclpy.init(args=args)                     # 开门前的准备
    node = GasSensorNode()                    # 装修 + 挂牌（构造）
    try: rclpy.spin(node)                   # 站到柜台后（运行）
    except KeyboardInterrupt: pass
    node.destroy_node(); rclpy.shutdown()     # 打烊摘牌（销毁）

就四块。但这四块之间有严格的先后依赖，顺序错了程序直接崩。下面这张图是它的完整生命周期，以及每一步"为什么必须在前一步之后"：

四个阶段里 只有 ③ 是"活着"：①②是开门前的准备，④是离场。图右那列红色文字全是真机上会碰到的报错——记住它们，排障能省一小时。

问题一：程序怎么知道自己"该是谁"

回到那行 super().__init__('gas_sensor_node')。Node 这个名字，是租来的，不是天生的。

你写个普通 Python 类，它就只是个对象，跑完就没了，谁也不知道它存在。只有继承 Node 并把这行 super().__init__() 跑完，这个对象才会向 ROS2 底层报到，从此有了全局身份、有了可被寻址的名字，出现在 ros2 node list 里。

问题二：节点到底"在哪"被观察到

报到完了，别人怎么看见它？靠两条命令——它们读的是同一张全局表（ROS2 图），所以只要节点在 spin，随时都能查到：

ros2 node list                  # 点名：现在都有谁在？→ /gas_sensor_node
ros2 node info /gas_sensor_node # 查户口：它发什么、收什么、提供哪些服务
#  Publishers: /gas_sensor [std_msgs/msg/Float32]   ← 构造函数第 ③ 步开的出口
#  Subscribers: /parameter_events, /rosout         ← ROS2 给每个节点自动挂的两个"系统话题"
#  Services: /gas_sensor_node/describe_parameters … ← 参数能力（EXP-10 细讲）

顺带解释一个新手常见疑惑：我明明没订阅 /parameter_events 和 /rosout，它们哪来的？——这是 ROS2 在创建节点时自动挂上的公共设施：前者让"改参数"这件事能广播出去，后者把日志汇聚到统一地方。所以每个节点天生就带参数与日志能力，不用你自己写。

问题三：spin 到底在转什么

这是本课最难也最关键的一步。看代码 rclpy.spin(node)，很容易以为它在"死循环占着不动"——恰恰相反，它是唯一让活能干起来的东西。

同一家店、同样几个窗口——区别只在于"老板在不在大堂"。这就是 spin 的全部含义。spin 不产生数据，它只负责"把已经到达的活派出去"。

问题四：谁在决定"先干哪个活"

到这儿有个新问题：如果同一时刻闹钟响了、消息也到了，谁先？这就引出第三个概念 Executor（执行器）。

代码怎么写：三种执行器的启动方式

上面三张卡片说的是"是什么"，这里给出"怎么写"。三者是层层递进的关系：默认单线程 → 换多线程 → 用回调组精细控制。先看清 main() 里那几行差在哪。

① 默认写法：rclpy.spin(node)（本课用这个）

# 最短的 main()：rclpy.spin 内部就是用 SingleThreadedExecutor 帮你 spin
def main():
    rclpy.init()
    node = GasSensorNode()
    try:
        rclpy.spin(node)          # 单线程，一次一个回调，其余排队
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

✔ 等价于下面这段——理解这点很重要，rclpy.spin() 不是另一种机制，只是"帮你 new 了一个单线程执行器并 spin 它"：

# 展开后的等价写法，看得更清楚
from rclpy.executors import SingleThreadedExecutor

executor = SingleThreadedExecutor()
executor.add_node(node)
try:
    executor.spin()
finally:
    executor.shutdown()
    node.destroy_node()

② 多线程执行器：让不同回调真正并行

把 SingleThreadedExecutor 换成 MultiThreadedExecutor，并指定线程数。注意：必须自己加锁，否则两个线程同时改一个变量就会出脏数据。

# 多线程版：num_threads 决定最多几个回调同时跑
import threading
import rclpy
from rclpy.node import Node
from std_msgs.msg import Float32
from rclpy.executors import MultiThreadedExecutor

class GasSensorNode(Node):
    def __init__(self):
        super().__init__('gas_sensor_node')
        # 共享数据必须先配一把锁
        self._lock = threading.Lock()
        self._last_ppm = 0.0
        self.create_timer(1.0, self.timer_cb)
        self.create_subscription(Float32, '/cmd', self.cmd_cb, 10)

    def timer_cb(self):
        ppm = read_mq5_once()
        with self._lock:              # 加锁写
            self._last_ppm = ppm

    def cmd_cb(self, msg):
        with self._lock:              # 加锁读
            print(f'当前 {self._last_ppm}')


def main():
    rclpy.init()
    node = GasSensorNode()

    # 关键一行：换成多线程执行器，开 4 个线程
    executor = MultiThreadedExecutor(num_threads=4)
    executor.add_node(node)

    try:
        executor.spin()             # 4 个线程同时从队列里取活干
    except KeyboardInterrupt:
        pass
    finally:
        executor.shutdown()
        node.destroy_node()
        rclpy.shutdown()

③ 回调组：精细控制"哪些能并行"

多线程执行器是"整体放开"，回调组是"按组放开"——你要明确告诉 ROS2：这几个回调可以并行，那几个必须排队。用法是在创建定时器/订阅时传入 callback_group 参数。

# 回调组：给回调分组，决定组内是"互斥"还是"可重入"
import time
import rclpy
from rclpy.node import Node
from std_msgs.msg import Float32
from rclpy.callback_groups import (
    MutuallyExclusiveCallbackGroup,   # 同组互斥：一次只跑一个
    ReentrantCallbackGroup,           # 同组可重入：可并发跑多个
)
from rclpy.executors import MultiThreadedExecutor


class RobotNode(Node):
    def __init__(self):
        super().__init__('robot_node')

        # 组 A：慢活，互斥——同一时刻只允许一个在跑
        self.group_slow = MutuallyExclusiveCallbackGroup()

        # 组 B：快活，可重入——多个可以同时跑
        self.group_fast = ReentrantCallbackGroup()

        # 创建时把组传进去，这个回调就归到该组
        self.create_timer(5.0, self.slow_cb, callback_group=self.group_slow)
        self.create_subscription(
            Float32, '/cmd', self.fast_cb, 10,
            callback_group=self.group_fast)      # ← 关键参数

    def slow_cb(self):
        time.sleep(3)               # 慢：即便多个触发，同组互斥也不会叠加

    def fast_cb(self, msg):
        print(msg.data)            # 快：多条消息可并发处理


def main():
    rclpy.init()
    node = RobotNode()

    # 回调组必须配合多线程执行器才有多线程效果
    executor = MultiThreadedExecutor(num_threads=4)
    executor.add_node(node)
    try:
        executor.spin()
    except KeyboardInterrupt:
        pass
    finally:
        executor.shutdown()
        node.destroy_node()
        rclpy.shutdown()

问题五：节点之间怎么"接上头"

现在 gas_sensor_node 已经能在柜台后正常干活了。可它发出的 /gas_sensor，怎么保证被 gas_monitor 收到？代码里从来没写过对方的 IP 或端口——这层事完全由 DDS 代办：

注意图里最关键的一句："发现层只牵线、不中转"。所以 ROS2 没有注册中心，也就不存在"注册中心一挂、全网瘫痪"的单点故障——这是 EXP-01 里 ROS1 vs ROS2 对比的底层原因。

问题六：三种"对话方式"怎么选

节点建好连上之后，说话的方式有三种。本课只需要建立"选型直觉"，细节留给后面三课：

这张图不用背，但要能自己画出来。EXP-06~08 会把每个分支掰开讲透。

03节点 API 速查手册

这一节不是"讲解"，是手册。写节点时你会反复回来查：调用哪个、传什么参数、返回什么、写错了会怎样。下面按"你在什么场景下要用"分组，每条都给签名 + 参数 + 返回 + 最小示例。

A. 程序入口：三件套（每个节点都必须有）

B. 创建节点：继承 vs 直接用

C. 通信：开出口 / 开入口

D. 参数：让节点"可调"

E. 日志：别用 print

F. 自省：我是谁、我周围有谁

G. 命名规则：为什么你的话题"跑"到了别处

ROS2 的名字有三种写法，写错一种，话题就会挂到你没预期的位置——这是"代码没错、就是收不到数据"的第二大原因。

# 相对名 + 命名空间：同一份代码跑两台车，互不串台
ros2 run my_pkg my_node --ros-args -r __ns:=/r1   # 话题变 /r1/cmd_vel
ros2 run my_pkg my_node --ros-args -r __ns:=/r2   # 话题变 /r2/cmd_vel
# 若代码里写死全局名 /cmd_vel，上面这条就失效——两台车照样抢同一个话题

04实操 A：编译并运行 gas_sensor_node

本实操不自己写新节点——直接跑现成的 gas_sensor_node，把 02 节讲的"构造 → spin → 销毁"在终端里一条条对上号。四条命令一组、共四步，一步步来。

source /opt/ros/humble/setup.bash
echo $ROS_DISTRO      # 打印 humble 才算加载成功

cd ~/ros2_ws
colcon build --packages-select ros_sensor ros_sensor_interfaces

source install/setup.bash
ros2 pkg executables ros_sensor   # 应列出 gas_sensor_node 等可执行名

cd ~/ros2_ws/src/xflz_eai_car1/driver/ros_sensor/ros_sensor/ros_sensor
cat gas_sensor_node.py

grep -A6 console_scripts setup.py

entry_points={
    'console_scripts': [
        'gas_sensor_node = ros_sensor.gas_sensor_node:main',
    ],
},

cd ~/ros2_ws
source /opt/ros/humble/setup.bash
source ~/ros2_ws/install/setup.bash
ros2 run ros_sensor gas_sensor_node

ros2 node list                     # 点名 → /gas_sensor_node
ros2 node info /gas_sensor_node    # 查户口：看 Publishers 段
ros2 topic list                    # → /gas_sensor、/parameter_events、/rosout

# 终端 2：Ctrl+C 停掉节点，改完名字后重新编译 + source
cd ~/ros2_ws
colcon build --packages-select ros_sensor
source install/setup.bash
ros2 run ros_sensor gas_sensor_node
# 终端 3：确认名字变了
ros2 node list        # → /my_gas_node

# 再开一个终端起第二个实例：名字被自动改成 _1
ros2 run ros_sensor gas_sensor_node
ros2 node list
# → /gas_sensor_node 和 /gas_sensor_node_1   （只给 WARN，不报错）

# 换命名空间：名字带上前辍，互不串台
ros2 run ros_sensor gas_sensor_node --ros-args -r __ns:=/r1
ros2 node list        # → /r1/gas_sensor_node

# 终端 2：按 Ctrl+C，应干净回到提示符（不该有红色 traceback）
# 终端 3：立刻再查一次
ros2 node list        # → 空（或已无 gas_sensor_node）

05实操 B：三节点组网与身份切换

# 终端 1–2 各起一个节点：talker 与 listener
ros2 run demo_nodes_cpp talker
ros2 run demo_nodes_cpp listener

# 全景观察
ros2 node list
ros2 node info /listener
#  Subscriber /parameter_events: [rcl_interfaces/msg/ParameterEvent]
#  Service /listener/get_parameters: [rcl_interfaces/srv/GetParameters]

# 用命名空间再造一个"平行世界"
ros2 run demo_nodes_cpp talker --ros-args -r __ns:=/r2
ros2 node list   # 共存：/talker 与 /r2/talker，互不串线

观察要点：/parameter_events 是每个节点都天然订阅的系统话题——这就是"节点天生带参数能力"的证据。

06注意事项与验收清单

07本课总结与思考题