3. 话题通信基础

任务目标

在ROS 2中创建两个独立节点：一个发布节点周期性发布字符串消息到话题 chatter，一个订阅节点接收并打印该消息。通过运行两个节点，验证话题通信机制。

环境依赖

• ROS 2（Humble或更新版本）已安装并source
• Python 3.8+
• 工作空间已创建（如 ~/ros2_ws/src）

完整实现

创建功能包并编写两个节点脚本。以下为最小可复现实现：

# 创建工作空间与功能包（若已存在可跳过）
mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python --node-name talker talker_pkg
ros2 pkg create --build-type ament_python --node-name listener listener_pkg

发布节点（talker_pkg/talker_pkg/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 ROS 2: {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_pkg/listener_pkg/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={
    'console_scripts': [
        'talker = talker_pkg.talker:main',
        # listener包中：
        'listener = listener_pkg.listener:main',
    ],
},

运行与验证

1. 编译并source：

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

2. 终端1运行发布节点：

ros2 run talker_pkg talker

3. 终端2运行订阅节点：

ros2 run listener_pkg listener

通过条件：订阅终端每0.5秒打印一条 I heard: "Hello ROS 2: N"（N为递增数字），且发布终端同步打印 Publishing: "Hello ROS 2: N"，即证明话题通信成功。

常见问题

• 找不到包：确认已执行 source install/setup.bash 且包名拼写正确。
• 话题不匹配：两个节点必须使用相同话题名 chatter，可用 ros2 topic list 检查。
• QoS不兼容：本示例发布与订阅均使用默认QoS（depth=10），若修改需保持一致。