2. 发布订阅基础通信

任务目标

编写两个ROS 2节点：一个发布者周期性发送字符串消息，一个订阅者接收并打印日志，验证ROS 2基础通信机制。

环境依赖

• Ubuntu 22.04 + ROS 2 Humble（或其他已安装的ROS 2发行版）
• 已创建ROS 2工作空间（如~/ros2_ws/src）
• Python 3 + rclpy库

完整实现

在~/ros2_ws/src/下创建包talker_listener，包含以下两个节点文件：

# 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 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.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()

编译与运行

1. 创建包并添加依赖：

cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python talker_listener
# 将上述两个.py文件放入 talker_listener/talker_listener/ 目录
# 编辑 setup.py，在 entry_points 中添加：
# 'talker = talker_listener.talker:main',
# 'listener = talker_listener.listener:main',

2. 编译并source：

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

3. 运行（两个终端）：

# 终端1
ros2 run talker_listener talker
# 终端2
ros2 run talker_listener listener

验证方法

观察终端2的listener节点是否持续打印Received: "Hello ROS 2: N"（N为递增数字），且与终端1发布者打印的Publishing内容一致。若两个终端均正常输出且消息编号同步递增，则通信验证通过。

也可用命令行工具验证：

ros2 topic list          # 应看到 /chatter
ros2 topic echo /chatter # 应看到 String 消息内容

常见问题

• 找不到包：确认已执行source install/setup.bash且包名拼写正确。
• 消息不匹配：检查发布者与订阅者是否使用相同话题名chatter和消息类型std_msgs/String。
• 编译报错：确认setup.py中entry_points格式正确，且package.xml已声明对std_msgs和rclpy的依赖。