4、话题通信基础：Publisher与Subscriber编写、自定义消息类型

聊到ROS2，话题通信是绕不开的核心。说白了，机器人各个模块之间怎么说话，全靠它。我刚开始接触ROS2时，觉得这玩意儿不就是发个消息嘛，有啥难的？结果真上手写机械臂控制程序时，才发现坑不少。今天咱们就把Publisher和Subscriber的编写、还有自定义消息类型，一次性捋清楚。

4.1 话题通信的本质

话题通信，你可以理解成一个“发布-订阅”模式。发布者（Publisher）往一个话题上扔数据，订阅者（Subscriber）从同一个话题上取数据。双方不需要知道对方是谁，只需要知道话题名字和消息类型就行。

我个人习惯把话题比作一个广播电台。Publisher是电台主持人，Subscriber是听众。主持人只管播，听众只管听。至于谁在听、有多少人在听，主持人根本不关心。这种解耦设计，让ROS2的节点可以灵活组合。

4.2 编写一个Publisher节点

咱们直接上手写代码。假设我们要控制一个机械臂的关节角度，需要定期发布关节状态消息。ROS2里最常用的消息类型是 sensor_msgs/msg/JointState。

先看一个简单的Publisher示例，用Python写：

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState

class JointStatePublisher(Node):
    def __init__(self):
        super().__init__('joint_state_publisher')
        # 创建一个发布者，话题名为'/joint_states'，消息队列长度为10
        self.publisher_ = self.create_publisher(JointState, '/joint_states', 10)
        self.timer = self.create_timer(0.1, self.timer_callback)  # 10Hz
        self.get_logger().info('关节状态发布节点已启动')

    def timer_callback(self):
        msg = JointState()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.name = ['joint1', 'joint2', 'joint3']
        msg.position = [0.5, 1.2, -0.8]
        msg.velocity = [0.0, 0.0, 0.0]
        msg.effort = [0.0, 0.0, 0.0]
        self.publisher_.publish(msg)

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

if __name__ == '__main__':
    main()

4.3 编写一个Subscriber节点

有发布就得有接收。Subscriber的写法跟Publisher对称，但逻辑是反的——它等着数据来，然后处理。

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState

class JointStateSubscriber(Node):
    def __init__(self):
        super().__init__('joint_state_subscriber')
        self.subscription = self.create_subscription(
            JointState,
            '/joint_states',
            self.listener_callback,
            10
        )
        self.subscription  # 防止被垃圾回收
        self.get_logger().info('关节状态订阅节点已启动')

    def listener_callback(self, msg):
        self.get_logger().info(f'收到关节状态: {msg.name}')
        self.get_logger().info(f'位置: {msg.position}')
        # 在这里可以添加你的控制逻辑

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

嗯，这里要注意：回调函数里不要做耗时操作。我曾经在回调里写了个复杂的运动学解算，结果导致消息积压，系统直接卡死。回调函数应该轻量，复杂计算另开线程或使用定时器。

4.4 自定义消息类型

ROS2自带的消息类型很多，但实际项目中总有些特殊需求。比如我要传递机械臂的末端位姿加上一个抓取力阈值，标准消息里没有这种组合。这时候就得自己定义消息类型。

自定义消息的步骤很简单，但容易踩坑。我一步步说：

1. 创建消息包：在功能包目录下创建 msg 文件夹。
2. 编写.msg文件：定义字段名和类型。
3. 修改CMakeLists.txt和package.xml：添加消息生成依赖。
4. 编译：用colcon build生成对应的Python/C++代码。

举个例子，定义一个机械臂抓取指令消息：

# 文件: my_robot_interfaces/msg/GripperCommand.msg
# 自定义消息：抓取指令
std_msgs/Header header
string gripper_name      # 夹爪名称
float64 target_position  # 目标位置（米）
float64 target_force     # 目标力（牛顿）
bool enable              # 使能标志

然后在 CMakeLists.txt 里加上：

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/GripperCommand.msg"
  DEPENDENCIES std_msgs
)

编译后，你就可以像使用标准消息一样使用它了：

from my_robot_interfaces.msg import GripperCommand

msg = GripperCommand()
msg.gripper_name = 'left_gripper'
msg.target_position = 0.05
msg.target_force = 20.0
msg.enable = True

4.5 消息类型对比

为了让你快速选型，我把常用的消息类型整理了一下：

4.6 调试技巧：用命令行工具验证

写完了代码，怎么知道对不对？别急着跑程序，先用ROS2的命令行工具测一下。

我个人最常用的三板斧：

• ros2 topic list：查看当前所有活跃话题。
• ros2 topic echo /topic_name：实时打印话题数据，看格式对不对。
• ros2 topic info /topic_name：查看话题的消息类型和发布者/订阅者数量。

比如，你启动Publisher节点后，在终端输入：

ros2 topic echo /joint_states

如果看到数据刷刷地往外蹦，说明发布成功了。如果没反应，八成是话题名拼错了，或者消息类型不匹配。

ros2 topic pub /joint_states sensor_msgs/msg/JointState "{header: {stamp: {sec: 0, nanosec: 0}}, name: ['j1'], position: [0.5]}"

4.7 常见问题与避坑

写话题通信时，有几个坑我几乎每次培训都会遇到：

• 话题名不一致：Publisher和Subscriber的话题名必须完全一样，大小写敏感。我见过有人把/joint_states写成/joint_state，找了一下午bug。
• 消息队列长度：create_publisher的第三个参数是队列长度。设太小容易丢消息，设太大占内存。机械臂控制一般10~20就够了。
• 回调函数阻塞：前面说过，回调里别做耗时操作。如果必须做，用MultiThreadedExecutor或者另起线程。
• 自定义消息未编译：改了.msg文件后，一定要重新编译整个工作空间。只编译单个包有时候不生效。