2. ROS2话题通信基础

任务目标

本案例将实现一个ROS2话题通信的最小闭环：创建自定义消息类型，编写一个发布者节点周期性发送消息，编写一个订阅者节点接收并打印消息，最终通过命令行工具验证消息传递的实时性。

环境准备

需要已安装ROS2（Humble或更高版本），并完成环境变量加载（source /opt/ros/humble/setup.bash）。创建工作空间并编译自定义消息接口包。

实现步骤

1. 创建自定义消息包

# 创建ROS2工作空间
mkdir -p ~/ros2_ws/src && cd ~/ros2_ws/src

# 创建自定义消息接口包
ros2 pkg create --build-type ament_cmake tutorial_interfaces

# 创建msg目录并定义消息文件
mkdir -p tutorial_interfaces/msg
cat > tutorial_interfaces/msg/Student.msg << 'EOF'
string name
int32 age
float32 score
EOF

编辑tutorial_interfaces/CMakeLists.txt，在find_package部分添加rosidl_default_generators依赖，并在文件末尾添加消息生成规则：

find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Student.msg"
)

编辑tutorial_interfaces/package.xml，添加以下依赖声明：

<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>

2. 创建发布者与订阅者节点

在~/ros2_ws/src下创建功能包：

ros2 pkg create --build-type ament_python py_talker_listener

创建发布者节点py_talker_listener/py_talker_listener/talker.py：

import rclpy
from rclpy.node import Node
from tutorial_interfaces.msg import Student

class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.publisher = self.create_publisher(Student, 'student_info', 10)
        self.timer = self.create_timer(0.5, self.timer_callback)  # 2Hz
        self.count = 0

    def timer_callback(self):
        msg = Student()
        msg.name = f'Student_{self.count}'
        msg.age = 20 + self.count % 10
        msg.score = 80.0 + self.count % 20
        self.publisher.publish(msg)
        self.get_logger().info(f'发布: {msg.name}, 年龄: {msg.age}, 成绩: {msg.score}')
        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()

创建订阅者节点py_talker_listener/py_talker_listener/listener.py：

import rclpy
from rclpy.node import Node
from tutorial_interfaces.msg import Student

class Listener(Node):
    def __init__(self):
        super().__init__('listener')
        self.subscription = self.create_subscription(
            Student, 'student_info', self.listener_callback, 10)
        self.receive_count = 0
        self.last_receive_time = self.get_clock().now()

    def listener_callback(self, msg):
        now = self.get_clock().now()
        interval = (now - self.last_receive_time).nanoseconds / 1e9
        self.last_receive_time = now
        self.receive_count += 1
        self.get_logger().info(
            f'收到: {msg.name}, 年龄: {msg.age}, 成绩: {msg.score:.1f}, '
            f'接收间隔: {interval:.3f}s, 累计: {self.receive_count}')

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

编辑py_talker_listener/setup.py，在entry_points中添加：

entry_points={
    'console_scripts': [
        'talker = py_talker_listener.talker:main',
        'listener = py_talker_listener.listener:main',
    ],
},

编辑py_talker_listener/package.xml，添加依赖：

<exec_depend>rclpy</exec_depend>
<exec_depend>tutorial_interfaces</exec_depend>

3. 编译与运行

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

# 终端1：运行发布者
ros2 run py_talker_listener talker

# 终端2：运行订阅者
ros2 run py_talker_listener listener

验证方法

订阅者终端应持续打印接收到的消息，且接收间隔稳定在0.5秒左右（与发布频率2Hz对应），说明消息实时传递。同时可打开第三个终端，使用ros2 topic echo /student_info查看话题数据，或使用ros2 topic hz /student_info查看实际发布频率。

预期输出示例

# 订阅者终端
[INFO] 收到: Student_0, 年龄: 20, 成绩: 80.0, 接收间隔: 0.502s, 累计: 1
[INFO] 收到: Student_1, 年龄: 21, 成绩: 81.0, 接收间隔: 0.498s, 累计: 2
[INFO] 收到: Student_2, 年龄: 22, 成绩: 82.0, 接收间隔: 0.501s, 累计: 3

通过条件：订阅者持续收到消息且间隔接近0.5秒，无报错，即验证通过。