4、订阅者节点实战：用Python写一个会“听”的节点

好，咱们接着聊。上一节我们写了个发布者，一个劲儿地往外发消息。那谁来收呢？当然就是订阅者了。这一节，我们就来亲手写一个Python订阅者节点，看看它是怎么“听到”消息的。

说白了，订阅者就是ROS2里的“听众”。它订阅某个话题，一旦有消息发布出来，它就会被唤醒，然后执行我们预先写好的回调函数。这个机制，是整个ROS2通信的灵魂。

4.1 先搭个架子：最简单的订阅者

我个人习惯，写任何节点之前，先把框架搭好。就像盖房子，先立柱子，再填砖。来看代码：

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__('simple_subscriber')
        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.listener_callback,
            10
        )
        self.subscription  # 防止被垃圾回收

    def listener_callback(self, msg):
        self.get_logger().info('收到消息: "%s"' % msg.data)

def main(args=None):
    rclpy.init(args=args)
    simple_subscriber = SimpleSubscriber()
    rclpy.spin(simple_subscriber)
    simple_subscriber.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

嗯，这段代码看着眼熟吧？跟发布者很像，但有个关键区别——create_subscription。它需要四个参数：消息类型、话题名、回调函数、QoS设置。

你想想看，回调函数就是那个listener_callback。一旦有消息到达，ROS2就会自动调用它。我在项目中遇到过，很多新手会在这里犯迷糊：回调函数里能不能做耗时操作？我的建议是——尽量别。回调函数应该轻量、快速，否则会阻塞整个节点的消息处理。

4.2 回调函数机制：到底是怎么工作的？

为什么会这样？回调函数到底是怎么被调用的？

其实，ROS2底层维护了一个事件循环。当你调用rclpy.spin()时，节点就进入了这个循环。它不停地检查有没有新消息到达，如果有，就立刻调用对应的回调函数。

我曾经调试过一个奇怪的问题：订阅者明明收到了消息，但回调函数就是没执行。查了半天，发现是rclpy.spin()被放在了另一个线程里，而主线程已经退出了。嗯，这里要注意——spin()必须在一个活着的线程里运行，否则节点就“死”了。

回调函数的执行顺序呢？默认是先进先出。但如果你设置了不同的QoS，顺序可能会变。我们后面会细说。

4.3 消息过滤：不是所有消息都要收

实际项目中，话题上的消息可能很密集。比如一个激光雷达，每秒发几千条消息。你的订阅者如果全收，CPU可能直接爆表。

那怎么办？消息过滤就派上用场了。ROS2提供了几种过滤方式：

• 时间戳过滤：只接收某个时间窗口内的消息
• 内容过滤：只接收满足特定条件的消息（ROS2 Humble及以上版本支持）
• 频率过滤：通过QoS设置，控制接收频率

我个人最常用的是频率过滤。比如，一个话题每秒发100条消息，但我只需要每秒处理10条。那我就在QoS里设置一个合适的队列深度，让ROS2自动帮我丢弃多余的消息。

def listener_callback(self, msg):
    if msg.data.startswith('重要'):
        self.process(msg)
    else:
        self.get_logger().debug('忽略非重要消息')
  

4.4 QoS设置：让你的订阅者更“聪明”

QoS，全称Quality of Service，服务质量。说白了，就是告诉ROS2：你希望怎么收消息？

ROS2的QoS设置比ROS1灵活得多。它有几个关键参数：

我在项目中遇到过这样一个场景：一个传感器节点发布数据，但订阅者启动得比较晚。如果使用默认的QoS（VOLATILE），订阅者就收不到之前发布的消息。后来我把durability改成TRANSIENT_LOCAL，问题就解决了。

你想想看，这就是QoS的威力。它不只是“快”或“慢”的问题，而是决定了节点之间如何“握手”。

4.5 实战：写一个带QoS的订阅者

光说不练假把式。我们来写一个稍微复杂点的订阅者，手动设置QoS：

from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy

class AdvancedSubscriber(Node):
    def __init__(self):
        super().__init__('advanced_subscriber')
        
        # 自定义QoS
        qos_profile = QoSProfile(
            history=HistoryPolicy.KEEP_LAST,
            depth=20,
            reliability=ReliabilityPolicy.BEST_EFFORT
        )
        
        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.listener_callback,
            qos_profile
        )

    def listener_callback(self, msg):
        self.get_logger().info('收到: %s' % msg.data)

看到没？我把reliability设成了BEST_EFFORT。这意味着如果网络不好，丢几条消息也无所谓。适合传感器数据这种“丢了就丢了”的场景。

但如果你做的是控制指令，那就得用RELIABLE。我曾经因为用错了QoS，导致机器人收到残缺的控制指令，差点撞墙。嗯，从那以后，我对QoS的选择就格外小心了。

4.6 知识体系：一张图看懂订阅者

说了这么多，我们来画张图，把订阅者的核心逻辑串起来：

这张图很清楚地展示了订阅者的工作流程：发布者往话题上发消息，订阅者通过QoS设置和消息过滤，决定接收哪些消息，然后触发回调函数进行处理。

4.7 避坑指南：我踩过的那些坑

最后，分享几个我实际踩过的坑，希望能帮你少走弯路：

• 回调函数里别sleep：我曾经在回调里加了time.sleep(1)，结果整个节点卡死。记住，回调函数要快！
• QoS不匹配：发布者和订阅者的QoS必须兼容。比如一个用RELIABLE，一个用BEST_EFFORT，就可能收不到消息。
• 忘记spin：没有rclpy.spin()，节点就是个空壳，永远不会处理消息。
• 话题名拼写错误：这个最坑。发布者和订阅者的话题名差一个字母，就收不到消息。我至少在这上面浪费过半小时。

好了，订阅者节点就讲到这里。记住三个关键词：回调函数、消息过滤、QoS设置。掌握了它们，你就掌握了ROS2通信的“听”功。