AMR机械臂协同作业系统 - 第2节

2. ROS2基础回顾

2. ROS2基础回顾：节点、话题、服务、动作的通信机制，以及Launch文件与参数服务器

各位同学，欢迎来到第二章。在正式开始机械臂和导航的协同实战之前，我们得先把地基夯实。ROS2的通信机制，说白了就是机器人各个零件之间「说话」的方式。我见过太多同学，代码写得飞起，结果节点之间就是聊不上天，最后发现是话题名拼错了，或者服务类型没对上。

这一章，我会带着大家把ROS2的四大通信方式——节点、话题、服务、动作——彻底捋一遍。另外，Launch文件和参数服务器也是日常开发离不开的。嗯，咱们开始吧。

本章核心知识体系

ROS2 核心通信机制与工具 节点 (Node) 话题 (Topic) 服务 (Service) 动作 (Action) 发布/订阅 请求/响应 目标/反馈/结果 Launch 文件 参数服务器 命令行工具 图例： 节点 话题 服务 动作 工具

2.1 节点：ROS2的最小执行单元

节点，你可以把它理解成一个独立的「小机器人脑细胞」。每个节点只负责一件事，比如控制机械臂的关节、处理激光雷达数据、或者规划导航路径。我个人习惯把节点设计得尽量「专一」，一个节点只干一件事，这样出了问题也好排查。

在ROS2里，创建一个节点非常简单。用Python的话，几行代码就搞定了：

import rclpy
from rclpy.node import Node

class MyRobotNode(Node):
    def __init__(self):
        super().__init__('my_robot_node')
        self.get_logger().info('节点已启动！')

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

if __name__ == '__main__':
    main()

你看，核心就是继承 Node 类，然后给它起个名字。这里有个坑——节点名在整个ROS2网络里必须是唯一的。我曾经在一个项目里，两个节点不小心重名了，结果其中一个怎么都启动不了，日志里也没报错，折腾了我半天才发现。

小技巧： 用 ros2 node list 命令可以查看当前所有活跃的节点。如果发现节点没起来，先跑这个命令看看，能省不少排查时间。

2.2 话题：异步的发布-订阅模式

话题是ROS2里最常用的通信方式。它就像是一个「广播电台」——发布者只管往外发消息，订阅者只管收消息，双方不需要知道对方的存在。这种松耦合的设计，在机器人系统里特别实用。

举个例子，机械臂的关节状态数据，通常就是通过话题发布的。你想想看，可能有多个节点同时需要这个数据：运动规划节点要它、可视化节点要它、日志记录节点也要它。如果用点对点通信，代码会变得非常混乱。

发布话题的代码长这样：

from std_msgs.msg import String

class TalkerNode(Node):
    def __init__(self):
        super().__init__('talker')
        self.publisher_ = self.create_publisher(String, 'chatter', 10)
        timer_period = 0.5  # 秒
        self.timer = self.create_timer(timer_period, self.timer_callback)

    def timer_callback(self):
        msg = String()
        msg.data = 'Hello, ROS2!'
        self.publisher_.publish(msg)
        self.get_logger().info(f'发布: {msg.data}')

订阅端就更简单了：

class ListenerNode(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'收到: {msg.data}')

注意： 话题的 QoS（服务质量）设置非常关键。我记得有一次，机械臂的关节数据总是丢包，后来发现是发布端的 QoS 设置成了「尽力而为」，而订阅端要求「可靠传输」。两者不匹配，数据就丢了。建议在项目初期就统一好 QoS 策略。

2.3 服务：同步的请求-响应模式

话题适合持续的数据流，但有些场景你需要「问一句，答一句」。比如，你让机械臂移动到某个位置，然后等它告诉你「到位了」或者「失败了」。这时候就该用服务了。

服务是同步的——客户端发请求，服务器处理，然后返回响应。这期间客户端会阻塞等待。说白了，就是「你问我答，答完拉倒」。

定义一个服务接口：

# srv/MoveTo.srv
float64 x
float64 y
float64 z
---
bool success
string message

服务端实现：

from your_package.srv import MoveTo

class MoveServer(Node):
    def __init__(self):
        super().__init__('move_server')
        self.srv = self.create_service(MoveTo, 'move_to', self.move_callback)

    def move_callback(self, request, response):
        self.get_logger().info(f'收到移动请求: ({request.x}, {request.y}, {request.z})')
        # 这里执行实际的移动逻辑
        response.success = True
        response.message = '移动完成'
        return response

避坑指南： 服务调用是阻塞的，如果你的服务端处理时间很长（比如超过几秒），客户端可能会超时。我曾经在控制机械臂时，一个移动指令执行了5秒，结果客户端等不及直接报错了。解决方案有两个：要么把超时时间设长，要么改用动作（Action）。

2.4 动作：带反馈的异步任务

动作是ROS2里最「高级」的通信方式。它结合了话题和服务的优点——你可以发起一个目标（像服务），然后在执行过程中不断收到反馈（像话题），最后拿到结果。对于机械臂抓取这种耗时任务，动作简直是量身定做。

动作的流程是这样的：

  1. 客户端发送一个目标（Goal）
  2. 服务端开始执行，并定期发送反馈（Feedback）
  3. 执行完成后，服务端返回结果（Result）
  4. 客户端可以随时取消目标

定义一个动作接口：

# action/Grasp.action
# 目标
string object_name
---
# 结果
bool success
float64 grasp_force
---
# 反馈
float64 progress
string status

动作服务端的骨架代码：

from your_package.action import Grasp

class GraspActionServer(Node):
    def __init__(self):
        super().__init__('grasp_action_server')
        self._action_server = ActionServer(
            self, Grasp, 'grasp', self.execute_callback)

    async def execute_callback(self, goal_handle):
        self.get_logger().info(f'开始抓取: {goal_handle.request.object_name}')
        feedback_msg = Grasp.Feedback()
        feedback_msg.progress = 0.0
        
        # 模拟抓取过程
        for i in range(10):
            feedback_msg.progress = (i + 1) * 10.0
            feedback_msg.status = f'正在接近物体... {feedback_msg.progress}%'
            goal_handle.publish_feedback(feedback_msg)
            await asyncio.sleep(0.5)
        
        goal_handle.succeed()
        result = Grasp.Result()
        result.success = True
        result.grasp_force = 5.0
        return result

核心要点： 动作的反馈机制在实战中非常有用。比如在Nav2导航中，机器人可以一边走一边告诉你「还有多远」、「当前速度多少」。如果中途发现路径被挡住了，客户端可以随时取消任务，重新规划。这种灵活性，是服务和话题单独做不到的。

2.5 Launch文件：一键启动全家桶

一个机器人系统通常有十几个甚至几十个节点。如果每次都要手动开终端、敲命令，那效率也太低了。Launch文件就是用来解决这个问题的——它可以把所有节点、参数、配置一次性搞定。

ROS2的Launch文件用Python写，比ROS1的XML灵活得多。我个人习惯把Launch文件分成三层：

  • 底层： 启动硬件驱动节点（如机械臂的can总线节点）
  • 中间层： 启动算法节点（如运动规划、状态估计）
  • 顶层： 启动应用层节点（如用户交互、任务调度）

一个简单的Launch文件示例：

from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    return LaunchDescription([
        Node(
            package='robot_bringup',
            executable='arm_driver',
            name='arm_driver_node',
            output='screen'
        ),
        Node(
            package='robot_bringup',
            executable='nav2_bringup',
            name='nav2_node',
            parameters=[{'use_sim_time': True}]
        ),
    ])

经验之谈： 我建议把参数尽量写在YAML文件里，然后在Launch文件中引用。这样改参数时不用动Launch文件本身，维护起来清爽很多。比如：parameters=['config/arm_params.yaml']

2.6 参数服务器：全局的配置中心

参数服务器，说白了就是一个「全局字典」。所有节点都可以往里面存数据、取数据。适合存放那些不经常变化但又需要全局访问的配置，比如PID参数、最大速度、传感器阈值等。

在代码里操作参数：

# 声明参数并获取
self.declare_parameter('max_velocity', 1.0)
max_vel = self.get_parameter('max_velocity').value

# 动态修改参数（需要回调）
self.add_on_set_parameters_callback(self.param_callback)

从命令行设置参数：

ros2 param set /arm_driver_node max_velocity 0.5

注意： 参数服务器不是数据库，别往里面塞大数据。我见过有人把整个地图数据塞进参数服务器，结果系统直接卡死。参数只适合存「配置项」，数据量大的东西请用话题或者文件。

好了，这一章的内容就是这些。节点、话题、服务、动作，再加上Launch文件和参数服务器，这些就是ROS2通信的全部家当了。下一章我们会把这些知识串起来，开始搭建机械臂和导航的联合仿真环境。嗯，到时候见。

← 上一章 📖 返回目录 下一章 →