2. ROS2话题通信基础

任务目标

本案例将创建两个ROS2节点：一个发布者节点talker周期性发布字符串消息到chatter话题，一个订阅者节点listener订阅该话题并打印接收到的消息。最后通过rqt_graph验证节点间的通信关系。

环境准备

确保已安装ROS2（Humble或更高版本），并完成环境配置：

source /opt/ros/humble/setup.bash

完整实现代码

创建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(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'发布: {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'收到: {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()

编译与运行

在ROS2工作空间中创建包并运行：

# 创建工作空间（若已有可跳过）
mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src

# 创建Python包
ros2 pkg create --build-type ament_python py_topic_demo

# 将上述两个.py文件放入 py_topic_demo/py_topic_demo/ 目录
# 编辑 setup.py，在 entry_points 中添加：
# '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

验证方法与判定标准

按以下步骤验证案例是否成功：

1. 终端输出检查：发布者终端每秒输出一条“发布: Hello ROS2: N”（N为递增数字），订阅者终端同步输出“收到: Hello ROS2: N”，且两者计数一致。
2. 话题列表确认：另开终端执行ros2 topic list，应看到/chatter话题。
3. rqt_graph可视化：执行rqt_graph，在图形界面中应看到/talker节点（椭圆）通过箭头指向/chatter话题（矩形），再由/chatter指向/listener节点，形成完整的发布-订阅链路。

通过条件：上述三项检查全部满足，即发布与订阅终端输出同步递增、/chatter话题存在、rqt_graph中节点关系与上图一致，即可判定案例实现成功。