2. ROS2话题通信基础

任务目标

在ROS2中创建两个Python节点：talker发布字符串消息到话题chatter，listener订阅该话题并打印接收到的内容。通过运行这两个节点，验证话题通信链路是否打通。

环境准备

本案例基于ROS2 Humble版本，需要预先安装rclpy库（ROS2 Python客户端库，随ROS2默认安装）。创建一个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/目录下创建两个节点文件。先创建发布者节点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(1.0, 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'I heard: "{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中添加节点入口：

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

验证方法：观察终端2的listener是否持续打印I heard: "Hello ROS2: N"（N为递增数字），同时终端1的talker打印对应的发布日志。若两个终端输出内容一致且数字同步递增，说明话题通信正常。也可在第三个终端执行ros2 topic echo /chatter查看话题数据流，或使用ros2 topic list确认/chatter话题存在。

预期输出示例

终端2（listener）应显示类似内容：

[INFO] [listener]: I heard: "Hello ROS2: 0"
[INFO] [listener]: I heard: "Hello ROS2: 1"
[INFO] [listener]: I heard: "Hello ROS2: 2"

以上输出由代码逻辑直接推出：发布者每秒递增计数并发布消息，订阅者收到后打印。实际运行时应观察到持续递增的输出流。