4. 话题消息类型自定义与使用

任务目标

在 ROS2 中，系统内置消息类型（如 std_msgs/String）无法满足复杂数据传输需求时，需要自定义消息类型。本案例将定义一个包含 int32、string 和 float64[] 三种字段的消息类型 SensorData.msg，并分别创建发布者与订阅者节点，完成该自定义消息的发送与接收。

环境准备

本案例基于 ROS2 Humble 版本，需要预先创建 ROS2 工作空间。以下命令在终端中执行：

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_cmake custom_msg_pkg
ros2 pkg create --build-type ament_python custom_msg_publisher
ros2 pkg create --build-type ament_python custom_msg_subscriber

实现步骤

1. 定义自定义消息类型

在 custom_msg_pkg 包中创建消息定义文件。首先创建 msg 目录并添加 SensorData.msg：

cd ~/ros2_ws/src/custom_msg_pkg
mkdir msg
cat > msg/SensorData.msg << EOF
int32 device_id
string device_name
float64[] readings
EOF

编辑 CMakeLists.txt，添加消息编译依赖：

find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/SensorData.msg"
)
ament_export_dependencies(rosidl_default_runtime)

编辑 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. 创建发布者节点

编辑 custom_msg_publisher/custom_msg_publisher/publisher_node.py：

import rclpy
from rclpy.node import Node
from custom_msg_pkg.msg import SensorData
import random

class SensorPublisher(Node):
    def __init__(self):
        super().__init__('sensor_publisher')
        self.publisher = self.create_publisher(SensorData, 'sensor_topic', 10)
        self.timer = self.create_timer(1.0, self.publish_data)

    def publish_data(self):
        msg = SensorData()
        msg.device_id = 1
        msg.device_name = 'temperature_sensor'
        msg.readings = [20.5 + random.uniform(-1, 1) for _ in range(3)]
        self.publisher.publish(msg)
        self.get_logger().info(f'Published: id={msg.device_id}, name={msg.device_name}, readings={msg.readings}')

def main(args=None):
    rclpy.init(args=args)
    node = SensorPublisher()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

编辑 setup.py，添加入口点：

entry_points={
    'console_scripts': [
        'sensor_publisher = custom_msg_publisher.publisher_node:main',
    ],
},

3. 创建订阅者节点

编辑 custom_msg_subscriber/custom_msg_subscriber/subscriber_node.py：

import rclpy
from rclpy.node import Node
from custom_msg_pkg.msg import SensorData

class SensorSubscriber(Node):
    def __init__(self):
        super().__init__('sensor_subscriber')
        self.subscription = self.create_subscription(
            SensorData, 'sensor_topic', self.listener_callback, 10)

    def listener_callback(self, msg):
        self.get_logger().info(
            f'Received: id={msg.device_id}, name={msg.device_name}, '
            f'readings={list(msg.readings)}')

def main(args=None):
    rclpy.init(args=args)
    node = SensorSubscriber()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

编辑 setup.py，添加入口点：

entry_points={
    'console_scripts': [
        'sensor_subscriber = custom_msg_subscriber.subscriber_node:main',
    ],
},

4. 添加包间依赖

在两个 Python 包的 package.xml 中添加：

<exec_depend>custom_msg_pkg</exec_depend>
<exec_depend>rclpy</exec_depend>

编译与运行

cd ~/ros2_ws
colcon build --packages-select custom_msg_pkg custom_msg_publisher custom_msg_subscriber
source install/setup.bash

打开两个终端，分别运行：

# 终端1
ros2 run custom_msg_publisher sensor_publisher

# 终端2
ros2 run custom_msg_subscriber sensor_subscriber

验证方法

满足以下条件即判定案例通过：

1. 编译无报错，colcon build 成功生成三个包；
2. 发布者终端每秒输出一条包含 device_id=1、device_name=temperature_sensor 和三个浮点数的日志；
3. 订阅者终端同步输出相同 device_id、device_name 和 readings 数据，且数值与发布者一致；
4. 在第三个终端执行 ros2 topic echo /sensor_topic，能看到完整的 SensorData 消息结构，字段名与 .msg 定义一致。

预期输出示例（订阅者终端）：

[INFO] Received: id=1, name=temperature_sensor, readings=[20.8, 19.9, 21.3]

常见问题

• 找不到自定义消息模块：确认已执行 source install/setup.bash，且 custom_msg_pkg 已成功编译；
• 消息字段类型不匹配：float64[] 在 Python 中对应 list，赋值时需使用列表；
• 话题无数据：检查发布者与订阅者是否使用相同话题名 sensor_topic，且两个节点均已正确启动。