【ROS 2 机器人技术】动作通信（Action）详解及机器人移动控制实战（附完整项目代码）

前言在机器人开发中我们经常需要执行一些耗时较长的任务比如让机器人导航到目标点、抓取物体、机械臂轨迹跟踪等。这些任务不仅要能发送目标还需要实时反馈进度并支持中途取消。ROS 2 中的动作Action通信机制正是为这种场景而生。本文将结合机器人专业背景从零开始讲解动作通信的原理并带你实现一个机器人移动控制的项目——服务器端模拟机器人移动过程客户端发送目标位姿并接收实时反馈。一、为什么需要动作通信ROS 2 提供了三种主要通信方式话题Topic单向流式传输适合传感器数据但无反馈和状态跟踪。服务Service一次请求一次响应适合快速计算但长时间任务会阻塞。动作Action专为长时间运行、可抢占、带反馈的任务设计。例如机器人导航到(5,5)中间需要不断报告当前位置且用户可能临时取消。用服务实现会超时用话题又缺乏请求-应答结构。动作通信完美解决了这个问题。二、动作通信的组成与工作原理动作通信基于客户端-服务器模型由三部分组成目标Goal客户端发送的任务指令。反馈Feedback服务器执行过程中定期发送的进度信息。结果Result任务完成后的最终响应。通信流程text客户端 服务器 | ---发送目标-------------- | | ---反馈循环---------- | | ---最终结果-------------- | | 可随时发送取消请求------- |内部实际依赖多个话题和服务实现但对开发者透明。三、项目实战控制机器人移动到目标位姿我们将创建一个名为robot_action_demo的功能包自定义一个动作接口MoveRobot.action包含目标目标位姿 (x, y, theta)反馈当前位姿和完成百分比结果是否成功及消息3.1 环境准备ROS 2 Humble其他版本类似Python 3.8colcon 构建工具3.2 创建功能包与自定义动作工作空间ws_action包结构如下textws_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.pyMoveRobot.action 定义text# 目标期望位姿 float64 target_x float64 target_y float64 target_theta --- # 结果是否成功 bool success string message --- # 反馈当前位置和进度 float64 current_x float64 current_y float32 progress_percent3.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] 发送目标: x5.00, y5.00, theta0.00 [反馈] 当前位姿: (0.50, 0.50), 进度 10.0% ... [反馈] 当前位姿: (4.50, 4.50), 进度 90.0% [结果] 成功到达目标!五、在机器人技术中的延伸真实机器人可将服务器中的模拟移动替换为运动控制指令通过里程计反馈真实位姿。导航ROS 2 Navigation2 框架的NavigateToPose就是一个标准动作接口。机械臂FollowJointTrajectory动作用于轨迹执行反馈各关节状态。掌握动作通信你就掌握了对机器人复杂行为的优雅控制模式。项目代码完整文件内容以下为robot_action_demo包的全部源码请按目录结构创建。1. 动作定义文件action/MoveRobot.actiontextfloat64 target_x float64 target_y float64 target_theta --- bool success string message --- float64 current_x float64 current_y float32 progress_percent2.package.xmlxml?xml version1.0? ?xml-model hrefhttp://download.ros.org/schema/package_format3.xsd schematypenshttp://www.w3.org/2001/XMLSchema? package format3 namerobot_action_demo/name version0.0.0/version descriptionDemo of ROS 2 action for robot movement/description maintainer emailstudentexample.comstudent/maintainer licenseApache-2.0/license buildtool_dependament_cmake/buildtool_depend buildtool_dependrosidl_default_generators/buildtool_depend dependrclpy/depend dependaction_msgs/depend dependbuiltin_interfaces/depend exec_dependrosidl_default_runtime/exec_depend member_of_grouprosidl_interface_packages/member_of_group export build_typeament_cmake/build_type /export /package3.CMakeLists.txtcmakecmake_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包安装)pythonfrom setuptools import setup package_name robot_action_demo setup( namepackage_name, version0.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_safeTrue, maintainerstudent, maintainer_emailstudentexample.com, descriptionROS 2 action demo for robot movement, licenseApache-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.cfgini[develop] script_dir$base/lib/robot_action_demo [install] install_scripts$base/lib/robot_action_demo6. 动作服务器robot_action_demo/move_robot_server.pypythonimport 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_callbackself.execute_callback, goal_callbackself.goal_callback, cancel_callbackself.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(successFalse, 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(successTrue, messagef成功到达 ({target_x:.2f}, {target_y:.2f})) self.get_logger().info(任务完成) return result def main(argsNone): rclpy.init(argsargs) server MoveRobotServer() # 使用多线程执行器以支持并发处理 executor MultiThreadedExecutor() rclpy.spin(server, executorexecutor) server.destroy() rclpy.shutdown() if __name__ __main__: main()7. 动作客户端robot_action_demo/move_robot_client.pypythonimport 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_groupReentrantCallbackGroup() ) self.get_logger().info(客户端已启动等待服务器...) def send_goal(self, x, y, theta0.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_sec5.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_callbackself.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(argsNone): rclpy.init(argsargs) 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九、总结本文从动作通信原理到实战带你实现了一个机器人移动控制的动作服务器与客户端。你可以在真实机器人平台上将模拟运动替换为速度指令和里程计反馈实现完整的闭环控制。掌握动作通信你的机器人编程能力将迈上新台阶。