2. ROS2话题通信基础

任务目标

本案例将创建两个ROS2节点：一个发布者节点周期性发布字符串消息到话题 chatter，一个订阅者节点订阅该话题并打印接收到的消息。通过运行这两个节点，验证ROS2话题通信机制的基本工作流程。

环境准备

需要已安装ROS2（推荐Humble或更高版本），并完成环境变量加载：

source /opt/ros/humble/setup.bash

完整实现

创建功能包并编写两个节点文件。以下代码为最小可运行实现，包含发布者与订阅者的完整逻辑：

# 创建功能包（在ros2_ws/src目录下执行）
# ros2 pkg create --build-type ament_python py_topic_demo

# 文件1: py_topic_demo/py_topic_demo/publisher_node.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class PublisherNode(Node):
    def __init__(self):
        super().__init__('publisher_node')
        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 = PublisherNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

# 文件2: py_topic_demo/py_topic_demo/subscriber_node.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SubscriberNode(Node):
    def __init__(self):
        super().__init__('subscriber_node')
        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 = SubscriberNode()
    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': [
        'publisher_node = py_topic_demo.publisher_node:main',
        'subscriber_node = py_topic_demo.subscriber_node:main',
    ],
},

编译与运行

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

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

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

验证方法

按以下步骤确认通信正常：

1. 发布者终端每秒输出一条 发布: Hello ROS2: N 日志，N从0递增；
2. 订阅者终端同步输出 收到: Hello ROS2: N，且N与发布者一致；
3. 另开终端执行 ros2 topic echo /chatter，可看到相同消息内容，证明话题确实在传输数据；
4. 执行 ros2 topic info /chatter 可看到Publisher count和Subscriber count均为1。

当订阅者终端持续打印与发布者编号一致的消息时，说明话题通信闭环验证通过。

常见问题

• 找不到功能包：确认已执行 source install/setup.bash，且编译时无报错。
• 消息收不到：检查两个节点是否在同一ROS2域（默认domain ID为0），且话题名称完全一致为 chatter。
• 编译报错：确认 setup.py 中entry_points格式正确，且包内 __init__.py 文件存在。