前言
在机器人开发中，我们经常需要执行一些耗时较长的任务，比如让机器人导航到目标点、抓取物体、机械臂轨迹跟踪等。这些任务不仅要能发送目标，还需要实时反馈进度，并支持中途取消。ROS 2 中的 动作（Action）通信机制 正是为这种场景而生。
本文将结合机器人专业背景，从零开始讲解动作通信的原理，并带你实现一个机器人移动控制的项目——服务器端模拟机器人移动过程，客户端发送目标位姿并接收实时反馈。

一、为什么需要动作通信？

ROS 2 提供了三种主要通信方式：

  • 话题（Topic）：单向流式传输，适合传感器数据，但无反馈和状态跟踪。

  • 服务（Service）：一次请求一次响应，适合快速计算，但长时间任务会阻塞。

  • 动作（Action）：专为长时间运行、可抢占、带反馈的任务设计。

例如机器人导航到(5,5)，中间需要不断报告当前位置，且用户可能临时取消。用服务实现会超时，用话题又缺乏请求-应答结构。动作通信完美解决了这个问题。

二、动作通信的组成与工作原理

动作通信基于客户端-服务器模型，由三部分组成：

  1. 目标（Goal）：客户端发送的任务指令。

  2. 反馈（Feedback）：服务器执行过程中定期发送的进度信息。

  3. 结果（Result）：任务完成后的最终响应。

通信流程：

text

客户端                       服务器
  | ---发送目标--------------> |
  | <---反馈（循环）---------- |
  | <---最终结果-------------- |
  | 可随时发送取消请求-------> |

内部实际依赖多个话题和服务实现，但对开发者透明。

三、项目实战：控制机器人移动到目标位姿

我们将创建一个名为 robot_action_demo 的功能包，自定义一个动作接口 MoveRobot.action，包含：

  • 目标：目标位姿 (x, y, theta)

  • 反馈：当前位姿和完成百分比

  • 结果：是否成功及消息

3.1 环境准备

  • ROS 2 Humble（其他版本类似）

  • Python 3.8+

  • colcon 构建工具

3.2 创建功能包与自定义动作

工作空间 ws_action，包结构如下：

text

ws_action/src/robot_action_demo/
├── action
│   └── MoveRobot.action
├── package.xml
├── setup.py
├── setup.cfg
└── robot_action_demo
    ├── __init__.py
    ├── move_robot_server.py
    └── move_robot_client.py

MoveRobot.action 定义：

text

# 目标：期望位姿
float64 target_x
float64 target_y
float64 target_theta
---
# 结果：是否成功
bool success
string message
---
# 反馈：当前位置和进度
float64 current_x
float64 current_y
float32 progress_percent
3.3 编写动作服务器

服务器模拟机器人从起点 (0,0,0) 移动到目标，每次更新 10%，发布反馈。

move_robot_server.py 关键代码逻辑：

  • 继承 Node 并创建 ActionServer

  • 回调函数执行任务：循环计算中间位姿，publish_feedback 发送，可检查是否被取消

  • 任务结束后返回结果

3.4 编写动作客户端

客户端发送目标，并异步接收反馈和结果。

move_robot_client.py 关键逻辑：

  • 创建 ActionClient

  • 等待服务器上线，发送目标

  • 设置反馈回调实时打印进度

  • 通过 get_result_async 获取最终结果

四、编译与运行

bash

# 在工作空间根目录
colcon build --packages-select robot_action_demo
source install/setup.bash

# 先启动服务器
ros2 run robot_action_demo move_robot_server

# 另开终端，运行客户端
ros2 run robot_action_demo move_robot_client --ros-args -p target_x:=5.0 -p target_y:=5.0

客户端输出示例：

text

[INFO] 发送目标: x=5.00, y=5.00, theta=0.00
[反馈] 当前位姿: (0.50, 0.50), 进度 10.0%
...
[反馈] 当前位姿: (4.50, 4.50), 进度 90.0%
[结果] 成功到达目标!

五、在机器人技术中的延伸

  • 真实机器人：可将服务器中的模拟移动替换为运动控制指令，通过里程计反馈真实位姿。

  • 导航：ROS 2 Navigation2 框架的 NavigateToPose 就是一个标准动作接口。

  • 机械臂：FollowJointTrajectory 动作用于轨迹执行，反馈各关节状态。

掌握动作通信，你就掌握了对机器人复杂行为的优雅控制模式。

项目代码（完整文件内容）

以下为 robot_action_demo 包的全部源码，请按目录结构创建。

1. 动作定义文件 action/MoveRobot.action

text

float64 target_x
float64 target_y
float64 target_theta
---
bool success
string message
---
float64 current_x
float64 current_y
float32 progress_percent

2. package.xml

xml

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>robot_action_demo</name>
  <version>0.0.0</version>
  <description>Demo of ROS 2 action for robot movement</description>
  <maintainer email="student@example.com">student</maintainer>
  <license>Apache-2.0</license>

  <buildtool_depend>ament_cmake</buildtool_depend>
  <buildtool_depend>rosidl_default_generators</buildtool_depend>

  <depend>rclpy</depend>
  <depend>action_msgs</depend>
  <depend>builtin_interfaces</depend>

  <exec_depend>rosidl_default_runtime</exec_depend>

  <member_of_group>rosidl_interface_packages</member_of_group>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

3. CMakeLists.txt

cmake

cmake_minimum_required(VERSION 3.8)
project(robot_action_demo)

if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
  add_compile_options(-Wall -Wextra -Wpedantic)
endif()

find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
find_package(rclpy REQUIRED)
find_package(action_msgs REQUIRED)
find_package(builtin_interfaces REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "action/MoveRobot.action"
  DEPENDENCIES builtin_interfaces action_msgs
)

ament_python_install_package(robot_action_demo)

install(
  DIRECTORY action
  DESTINATION share/${PROJECT_NAME}/action
)

install(
  PROGRAMS
    robot_action_demo/move_robot_server.py
    robot_action_demo/move_robot_client.py
  DESTINATION lib/${PROJECT_NAME}
)

ament_package()

4. setup.py (为了兼容，也可以只使用CMakeLists，此处保留Python包安装)

python

from setuptools import setup

package_name = 'robot_action_demo'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='student',
    maintainer_email='student@example.com',
    description='ROS 2 action demo for robot movement',
    license='Apache-2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'move_robot_server = robot_action_demo.move_robot_server:main',
            'move_robot_client = robot_action_demo.move_robot_client:main',
        ],
    },
)

5. setup.cfg

ini

[develop]
script_dir=$base/lib/robot_action_demo
[install]
install_scripts=$base/lib/robot_action_demo

6. 动作服务器 robot_action_demo/move_robot_server.py

python

import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer, GoalResponse
from rclpy.executors import MultiThreadedExecutor
import time
import math

# 导入自定义动作
from robot_action_demo.action import MoveRobot

class MoveRobotServer(Node):
    def __init__(self):
        super().__init__('move_robot_server')
        self._action_server = ActionServer(
            self,
            MoveRobot,
            'move_robot',
            execute_callback=self.execute_callback,
            goal_callback=self.goal_callback,
            cancel_callback=self.cancel_callback
        )
        self.get_logger().info('动作服务器已启动，等待目标...')

    def goal_callback(self, goal_request):
        """目标请求到来时调用，可用于校验目标合法性"""
        self.get_logger().info(f'收到目标: x={goal_request.target_x:.2f}, y={goal_request.target_y:.2f}')
        # 这里可添加合法性检查，例如范围限制
        if abs(goal_request.target_x) > 10.0 or abs(goal_request.target_y) > 10.0:
            self.get_logger().warn('目标超出范围，拒绝')
            return GoalResponse.REJECT
        return GoalResponse.ACCEPT

    def cancel_callback(self, goal_handle):
        """客户端请求取消时调用"""
        self.get_logger().info('收到取消请求')
        return True  # 接受取消

    async def execute_callback(self, goal_handle):
        """实际执行任务，可异步"""
        target_x = goal_handle.request.target_x
        target_y = goal_handle.request.target_y
        target_theta = goal_handle.request.target_theta

        self.get_logger().info(f'开始执行移动任务到 ({target_x:.2f}, {target_y:.2f})')

        # 模拟参数：起点(0,0)，总步数10，每步0.5秒
        steps = 10
        feedback_msg = MoveRobot.Feedback()

        for i in range(1, steps + 1):
            # 检查是否被取消
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                self.get_logger().info('任务被取消')
                return MoveRobot.Result(success=False, message='任务被取消')

            # 计算中间位姿
            progress = i / float(steps)
            current_x = target_x * progress
            current_y = target_y * progress
            feedback_msg.current_x = current_x
            feedback_msg.current_y = current_y
            feedback_msg.progress_percent = progress * 100.0

            # 发布反馈
            self.get_logger().info(f'发布反馈: ({current_x:.2f}, {current_y:.2f}), 进度 {progress*100:.1f}%')
            goal_handle.publish_feedback(feedback_msg)

            # 模拟移动耗时
            time.sleep(0.5)

        # 任务完成，返回成功结果
        goal_handle.succeed()
        result = MoveRobot.Result(success=True, message=f'成功到达 ({target_x:.2f}, {target_y:.2f})')
        self.get_logger().info('任务完成')
        return result

def main(args=None):
    rclpy.init(args=args)
    server = MoveRobotServer()
    # 使用多线程执行器以支持并发处理
    executor = MultiThreadedExecutor()
    rclpy.spin(server, executor=executor)
    server.destroy()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

7. 动作客户端 robot_action_demo/move_robot_client.py

python

import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from rclpy.callback_groups import ReentrantCallbackGroup

from robot_action_demo.action import MoveRobot

class MoveRobotClient(Node):
    def __init__(self):
        super().__init__('move_robot_client')
        self._action_client = ActionClient(
            self,
            MoveRobot,
            'move_robot',
            callback_group=ReentrantCallbackGroup()
        )
        self.get_logger().info('客户端已启动，等待服务器...')

    def send_goal(self, x, y, theta=0.0):
        """发送目标并注册回调"""
        goal_msg = MoveRobot.Goal()
        goal_msg.target_x = x
        goal_msg.target_y = y
        goal_msg.target_theta = theta

        # 等待服务器上线
        if not self._action_client.wait_for_server(timeout_sec=5.0):
            self.get_logger().error('动作服务器未上线，退出')
            return

        self.get_logger().info(f'发送目标: x={x:.2f}, y={y:.2f}, theta={theta:.2f}')

        # 异步发送目标
        send_goal_future = self._action_client.send_goal_async(
            goal_msg,
            feedback_callback=self.feedback_callback
        )
        send_goal_future.add_done_callback(self.goal_response_callback)

    def goal_response_callback(self, future):
        """目标请求被服务器接受/拒绝时的回调"""
        goal_handle = future.result()
        if not goal_handle.accepted:
            self.get_logger().error('目标被拒绝')
            return

        self.get_logger().info('目标被接受，等待结果...')
        # 获取结果
        result_future = goal_handle.get_result_async()
        result_future.add_done_callback(self.result_callback)

    def feedback_callback(self, feedback_msg):
        """处理反馈信息"""
        fb = feedback_msg.feedback
        self.get_logger().info(f'[反馈] 当前位姿: ({fb.current_x:.2f}, {fb.current_y:.2f}), '
                               f'进度 {fb.progress_percent:.1f}%')

    def result_callback(self, future):
        """获取最终结果"""
        result = future.result().result
        if result.success:
            self.get_logger().info(f'[结果] 成功: {result.message}')
        else:
            self.get_logger().info(f'[结果] 失败: {result.message}')
        rclpy.shutdown()

def main(args=None):
    rclpy.init(args=args)
    client = MoveRobotClient()
    # 从参数读取目标，默认(5,5)
    from rclpy.utilities import remove_ros_args
    import sys
    # 简单解析参数
    target_x = 5.0
    target_y = 5.0
    for arg in sys.argv:
        if arg.startswith('target_x:='):
            target_x = float(arg.split(':=')[1])
        if arg.startswith('target_y:='):
            target_y = float(arg.split(':=')[1])

    client.send_goal(target_x, target_y)
    rclpy.spin(client)

if __name__ == '__main__':
    main()

8. 资源注册文件（需手动创建）

在 robot_action_demo 包根目录下创建 resource 文件夹，里面新建一个名为 robot_action_demo 的空文件（与包名相同），用于标记包安装。

构建与运行

bash

# 回到工作空间根目录
cd ~/ws_action
colcon build --symlink-install
source install/setup.bash
# 运行服务器
ros2 run robot_action_demo move_robot_server
# 新终端运行客户端
ros2 run robot_action_demo move_robot_client --ros-args -p target_x:=8.0 -p target_y:=3.0

九、总结

本文从动作通信原理到实战，带你实现了一个机器人移动控制的动作服务器与客户端。你可以在真实机器人平台上，将模拟运动替换为速度指令和里程计反馈，实现完整的闭环控制。掌握动作通信，你的机器人编程能力将迈上新台阶。

# 机器人
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

TVA在具身智能技术演进中的独特价值（19）

avatar DAMO开发者矩阵
cover

TVA在具身智能系统演进中的独特价值（14）

avatar DAMO开发者矩阵
cover

TVA在具身智能技术演进中的独特价值（20）

avatar DAMO开发者矩阵