2. ROS2话题通信基础

任务目标

在ROS2中创建发布者节点和订阅者节点，实现字符串消息的实时单向传输。通过本案例理解话题通信的核心机制：节点通过话题解耦，发布者向话题写入消息，订阅者从话题读取消息，双方无需知道对方的存在。

环境准备

本案例基于ROS2 Humble版本，需要预先安装ROS2并完成环境配置。创建工作空间并创建功能包：

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python py_topic_demo

关键完整代码

在py_topic_demo/py_topic_demo/目录下创建两个Python文件，分别实现发布者和订阅者。

发布者节点（talker.py）：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.publisher = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(0.5, self.timer_callback)
        self.count = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello ROS2: {self.count}'
        self.publisher.publish(msg)
        self.get_logger().info(f'Publishing: "{msg.data}"')
        self.count += 1

def main(args=None):
    rclpy.init(args=args)
    node = Talker()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

订阅者节点（listener.py）：

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class Listener(Node):
    def __init__(self):
        super().__init__('listener')
        self.subscription = self.create_subscription(
            String, 'chatter', self.listener_callback, 10)

    def listener_callback(self, msg):
        self.get_logger().info(f'Received: "{msg.data}"')

def main(args=None):
    rclpy.init(args=args)
    node = Listener()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

配置入口点（setup.py）：

entry_points={
    'console_scripts': [
        'talker = py_topic_demo.talker:main',
        'listener = py_topic_demo.listener:main',
    ],
},

编译与运行

cd ~/ros2_ws
colcon build --packages-select py_topic_demo
source install/setup.bash

# 终端1：运行发布者
ros2 run py_topic_demo talker

# 终端2：运行订阅者
ros2 run py_topic_demo listener

验证方法与判定标准

运行后观察两个终端的输出：

• 发布者终端每0.5秒输出一条Publishing: "Hello ROS2: N"日志，N从0开始递增；
• 订阅者终端同步输出Received: "Hello ROS2: N"，且N与发布者一致；
• 在第三个终端执行ros2 topic list，应看到/chatter话题；执行ros2 topic echo /chatter，应能看到实时消息内容。

满足以上条件即说明话题通信链路正常，发布者与订阅者通过chatter话题成功实现了字符串消息的实时传输。

话题通信机制验证

先启动发布者，再启动订阅者，订阅者仍能收到后续消息，说明话题通信是异步解耦的。先关闭发布者，订阅者保持运行不报错；先关闭订阅者，发布者继续发布不中断，验证了节点间无直接依赖关系。

图中展示了本案例的通信架构：发布者节点talker将String类型消息发布到/chatter话题，订阅者节点listener从同一话题接收消息。两个节点之间没有直接连接，仅通过话题名称进行解耦通信，这正是ROS2话题通信的核心设计思想。