自定义话题消息：msg文件定义、编译规则、Python中调用自定义消息

好，咱们接着聊。上一章我们搞定了标准消息类型，但说实话，实际项目里你很快就会发现——标准消息根本不够用。

举个例子。我在做一个六轴机械臂的力控项目时，需要同时传递关节角度、末端位姿、力矩值、时间戳。用标准消息？要么拆成四五个话题发，要么自己拼字符串。两种方式都让人头疼。后来我学乖了——自己定义消息结构，一劳永逸。

为什么需要自定义消息？

说白了，标准消息是ROS官方给你搭的积木。但你的机器人不是积木，它有自己独特的数据结构。

• 数据聚合：把多个相关字段打包成一个整体
• 类型安全：编译时就能发现类型错误，不用等到运行时崩溃
• 代码可读性：看到消息名就知道里面装了什么
• 跨语言支持：C++、Python、Java都能用同一个msg文件

msg文件长什么样？

msg文件其实就是个文本文件，后缀是.msg。放在功能包里的msg文件夹下。语法极其简单：

# 这是注释，用#号开头
# 格式：字段类型 字段名

# 机械臂关节状态消息
string robot_name        # 机器人名字，比如"arm_01"
uint8 joint_count        # 关节数量
float32[] joint_angles   # 关节角度数组，单位弧度
float32[] joint_velocities # 关节速度数组
float32[] joint_torques  # 关节力矩数组
bool is_moving           # 是否正在运动
time stamp               # 时间戳

嗯，这里要注意：字段类型必须是ROS支持的基本类型。我列个表给你看：

类型	说明	C++对应	Python对应
bool	布尔值	uint8_t	bool
int8/uint8	8位整数	int8_t/uint8_t	int
int16/uint16	16位整数	int16_t/uint16_t	int
int32/uint32	32位整数	int32_t/uint32_t	int
float32	32位浮点	float	float
float64	64位浮点	double	float
string	字符串	std::string	str
time	ROS时间	ros::Time	rospy.Time
duration	ROS时长	ros::Duration	rospy.Duration
uint8[]	可变长数组	std::vector<uint8_t>	list

编译规则配置

msg文件写好了，但ROS还不知道它的存在。你得告诉编译系统：嘿，帮我生成代码！

首先，在CMakeLists.txt里做三件事：

# 1. 添加message_generation依赖
find_package(catkin REQUIRED COMPONENTS
  roscpp
  rospy
  std_msgs
  message_generation  # 这行是关键
)

# 2. 声明你要编译的msg文件
add_message_files(
  FILES
  JointState.msg
  # 如果有多个msg，继续往下写
  # EndEffectorPose.msg
)

# 3. 生成消息代码
generate_messages(
  DEPENDENCIES
  std_msgs  # 如果你的msg里用了std_msgs的类型
)

然后，在package.xml里也要声明依赖：

<build_depend>message_generation</build_depend>
<exec_depend>message_runtime</exec_depend>

我曾经犯过一个低级错误——只加了build_depend，忘了加exec_depend。结果编译通过，运行时却报找不到消息。排查了半天，最后发现是运行时依赖没加。嗯，这种坑踩过一次就记住了。

编译与验证

配置好之后，回到工作空间根目录：

cd ~/catkin_ws
catkin_make
# 或者用 catkin build，看你习惯

编译成功后，用rosmsg命令验证：

rosmsg show your_package_name/JointState

如果看到你定义的字段列表，说明消息生成成功了。我个人习惯每次改完msg都跑一下这个命令，确认没写错。

Python中调用自定义消息

消息编译好了，怎么在Python里用？其实和标准消息一模一样。

发布者示例

#!/usr/bin/env python3
import rospy
from your_package_name.msg import JointState

def publish_joint_state():
    pub = rospy.Publisher('joint_states', JointState, queue_size=10)
    rospy.init_node('joint_state_publisher', anonymous=True)
    rate = rospy.Rate(10)  # 10Hz

    while not rospy.is_shutdown():
        msg = JointState()
        msg.robot_name = "arm_01"
        msg.joint_count = 6
        msg.joint_angles = [0.0, -0.5, 1.2, 0.0, 0.8, 0.0]
        msg.joint_velocities = [0.1, 0.0, -0.2, 0.0, 0.0, 0.0]
        msg.joint_torques = [0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
        msg.is_moving = True
        msg.stamp = rospy.Time.now()

        pub.publish(msg)
        rate.sleep()

if __name__ == '__main__':
    try:
        publish_joint_state()
    except rospy.ROSInterruptException:
        pass

订阅者示例

#!/usr/bin/env python3
import rospy
from your_package_name.msg import JointState

def callback(msg):
    rospy.loginfo(f"收到 {msg.robot_name} 的状态")
    rospy.loginfo(f"关节角度: {msg.joint_angles}")
    rospy.loginfo(f"运动状态: {'运动中' if msg.is_moving else '静止'}")

def listen_joint_state():
    rospy.init_node('joint_state_listener', anonymous=True)
    rospy.Subscriber('joint_states', JointState, callback)
    rospy.spin()

if __name__ == '__main__':
    listen_joint_state()

消息嵌套与复用

有时候一个消息不够用，你可以在一个msg里引用另一个msg。比如：

# RobotArmState.msg
JointState arm_joints
float64[3] end_effector_pose
float64 gripper_force

前提是JointState.msg也在同一个包或者依赖包里。编译时ROS会自动处理这种嵌套关系。

本章知识体系

下面这张图帮你理清自定义消息的完整流程：

从定义到编译，再到Python调用，每一步都环环相扣。你可能会问：为什么非要这么麻烦？直接用字典传数据不行吗？

行，但不行。字典没有类型检查，你传个字符串进去，订阅方以为是浮点数，运行时才报错。自定义消息在编译阶段就能发现这种问题。我在一个协作机器人项目里吃过这个亏——调试了三天，最后发现是字段类型写错了。从那以后，我坚持用自定义消息，再也不用字典凑合了。

可可精品资料库