ROS2核心概念之动作

动作（Action）是ROS2中用于处理长时间运行任务的通信机制。与话题和服务不同，动作支持双向通信：客户端可以发送目标请求、接收反馈，并在任务完成时获取最终结果。本文将带你从零开始理解动作的概念，并通过完整代码示例掌握其用法。## 动作的基本结构动作由三部分组成：- 目标：客户端发送给服务端的请求（例如“移动到坐标(1,2)”）- 反馈：服务端在任务执行过程中定期发送的状态信息（例如“当前完成度50%”）- 结果：任务完成后返回的最终数据（例如“到达目标位置”）这种结构非常适合机器人导航、机械臂抓取等需要监控进度的场景。## 创建动作接口首先，我们需要定义动作的通信格式。ROS2使用.action文件来定义动作接口。创建一个名为MoveRobot.action的文件（放在action/目录下）：plaintext# 目标int32 target_xint32 target_y---# 结果bool successstring message---# 反馈int32 current_xint32 current_yfloat32 progress动作定义分三部分，用---分隔：1. 第一部分：目标数据结构2. 第二部分：结果数据结构3. 第三部分：反馈数据结构## 编写动作服务端动作服务端负责接收目标、执行任务并发送反馈。以下代码实现了一个模拟机器人移动的服务端。pythonimport rclpyfrom rclpy.action import ActionServerfrom rclpy.node import Nodefrom example_interfaces.action import Fibonacci # 使用ROS2内置Fibonacci动作示例class MoveActionServer(Node): def __init__(self): super().__init__('move_action_server') # 创建动作服务器，指定动作类型、动作名称和回调函数 self._action_server = ActionServer( self, Fibonacci, # 动作接口类型 'move_robot', # 动作名称 self.execute_callback) # 执行回调 self.get_logger().info('动作服务器已启动') def execute_callback(self, goal_handle): """处理动作目标的回调函数""" self.get_logger().info(f'接收目标: {goal_handle.request.order}') # 初始化反馈消息 feedback_msg = Fibonacci.Feedback() feedback_msg.sequence = [] # 模拟任务执行过程 for i in range(1, goal_handle.request.order + 1): # 检查是否被客户端取消 if goal_handle.is_cancel_requested: goal_handle.canceled() self.get_logger().info('目标被取消') return Fibonacci.Result() # 更新反馈 feedback_msg.sequence.append(i) self.get_logger().info(f'进度: {i}/{goal_handle.request.order}') goal_handle.publish_feedback(feedback_msg) # 模拟耗时操作 rclpy.spin_once(self, timeout_sec=0.5) # 设置最终结果 result = Fibonacci.Result() result.sequence = feedback_msg.sequence goal_handle.succeed() self.get_logger().info('任务完成') return resultdef main(args=None): rclpy.init(args=args) node = MoveActionServer() rclpy.spin(node) rclpy.shutdown()if __name__ == '__main__': main()关键点说明：- execute_callback 是核心处理函数，接收GoalHandle对象- 通过goal_handle.publish_feedback()发送进度反馈- 使用goal_handle.is_cancel_requested检查取消请求- 最终通过goal_handle.succeed()标记成功## 编写动作客户端动作客户端负责发送目标请求并处理反馈和结果。pythonimport rclpyfrom rclpy.action import ActionClientfrom rclpy.node import Nodefrom example_interfaces.action import Fibonacciclass MoveActionClient(Node): def __init__(self): super().__init__('move_action_client') # 创建动作客户端 self._action_client = ActionClient( self, Fibonacci, 'move_robot') self.get_logger().info('动作客户端已创建') def send_goal(self, order): """发送动作目标""" # 等待动作服务器就绪 while not self._action_client.wait_for_server(timeout_sec=1.0): self.get_logger().warn('等待动作服务器...') # 创建目标消息 goal_msg = Fibonacci.Goal() goal_msg.order = order # 发送目标并设置回调 self._send_goal_future = self._action_client.send_goal_async( goal_msg, feedback_callback=self.feedback_callback) # 绑定结果回调 self._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('目标被接受') # 等待最终结果 self._get_result_future = goal_handle.get_result_async() self._get_result_future.add_done_callback(self.get_result_callback) def get_result_callback(self, future): """处理最终结果的回调""" result = future.result().result self.get_logger().info(f'计算结果: {result.sequence}') rclpy.shutdown() def feedback_callback(self, feedback_msg): """处理进度反馈的回调""" feedback = feedback_msg.feedback self.get_logger().info(f'收到反馈: {feedback.sequence}')def main(args=None): rclpy.init(args=args) node = MoveActionClient() node.send_goal(10) # 请求计算斐波那契数列前10项 rclpy.spin(node)if __name__ == '__main__': main()客户端工作流程：1. 等待服务器就绪2. 发送目标并注册回调3. 服务器接受目标后，继续等待结果4. 通过feedback_callback实时接收进度5. 最终通过get_result_callback获取结果## 运行测试1. 首先编译动作接口（如果使用自定义动作）2. 在一个终端运行服务端： bash python3 move_action_server.py 3. 在另一个终端运行客户端： bash python3 move_action_client.py 你会看到服务端输出进度信息，客户端实时显示反馈，最终输出计算结果。## 高级用法：处理超时和取消在实际应用中，可能需要处理超时或手动取消任务。以下是一个增强版客户端示例：pythonimport timeimport rclpyfrom rclpy.action import ActionClientfrom rclpy.node import Nodefrom example_interfaces.action import Fibonacciclass AdvancedActionClient(Node): def __init__(self): super().__init__('advanced_action_client') self._action_client = ActionClient(self, Fibonacci, 'move_robot') self._goal_handle = None def send_goal_with_timeout(self, order, timeout=5.0): """发送带超时的目标""" goal_msg = Fibonacci.Goal() goal_msg.order = order # 发送目标 goal_future = self._action_client.send_goal_async(goal_msg) rclpy.spin_until_future_complete(self, goal_future, timeout_sec=2.0) if not goal_future.done(): self.get_logger().warn('服务器未响应，取消请求') return goal_handle = goal_future.result() if not goal_handle.accepted: return self._goal_handle = goal_handle # 启动定时器，如果超时则取消 self.create_timer(timeout, self.cancel_goal) # 获取结果 result_future = goal_handle.get_result_async() rclpy.spin_until_future_complete(self, result_future) # 取消定时器 self.destroy_timer(self._cancel_timer) return result_future.result().result def cancel_goal(self): """取消当前目标""" if self._goal_handle: cancel_future = self._goal_handle.cancel_goal_async() rclpy.spin_until_future_complete(self, cancel_future) self.get_logger().info('已发送取消请求')## 总结动作是ROS2中处理长时间运行任务的强大工具。通过本文的学习，你掌握了：1. 动作的三大核心组件：目标、反馈、结果2. 动作接口定义：使用.action文件描述数据结构3. 服务端实现：接收目标、发送反馈、返回结果4. 客户端实现：发送请求、处理反馈和结果5. 高级特性：超时控制、任务取消动作机制让机器人编程更加灵活：你可以像调用远程函数一样启动任务，同时实时监控进度，必要时还能取消任务。这为构建复杂的机器人系统提供了坚实的基础。在实际项目中，建议将动作接口定义放在单独的功能包中，便于复用。同时注意正确处理取消请求和异常情况，确保系统健壮性。掌握动作后，你将能够更高效地开发导航、机械臂控制等需要长时间交互的机器人应用。