目录1 action工作原理2 创建客户端3 创建服务端4 运行参考文档https://zhuanlan.zhihu.com/p/1913364862865740535action文件格式#目标 Goal float32 aim --- #结果 Result bool result --- #响应 Feedback float32 current1 action工作原理actionaction是之前msg和srv的组合至少2个server请求响应和1个topic发布订阅。AI说和还有一个隐藏的topic。其中步骤1和2构成一个请求响应用于确定是否提供服务。步骤2和步骤5构成一个请求响应用于反馈最终结果。步骤4构成一个topic用于实时反馈结果feedback。1 客户端向Goal Service服务端发送请求2 Goal Service服务端反馈目标接受状态接收或拒绝3 客户端向Result Service服务端发送请求4 动作服务器开始计算并通过Feedback Topic反馈中间结果5 最终结果计算完毕后Result Service服务端反馈结果给Result Service客户端 由以上的交互流程2 创建客户端共三个主要回调import rclpy from rclpy.action import ActionClient from rclpy.node import Node from pkg_interfaces.action import MyAction class ActionClientNode(Node): def __init__(self): super().__init__(action_client) # 实例化动作客户端不再通过函数创建 self._action_client ActionClient(self, MyAction, MyAction) def send_goal(self, aim): # 等待动作服务器就绪 while not self._action_client.wait_for_server(timeout_sec 1.0): self.get_logger().info(service not ready) # 构建目标消息 goal_msg MyAction.Goal() goal_msg.aim aim # 异步发送目标,请求对端是否提供服务 self._send_goal_future self._action_client.send_goal_async( goal_msg, # 对应 topic 的订阅回调 feedback_callbackself.feedback_callback) # feedback回调对应话题 # 绑定目标响应回调函数确认服务器是否接受目标 self._send_goal_future.add_done_callback(self.goal_response_callback) # 回调1 请求确定是否提供提供服务当确定提供服务后请求最终结果 def goal_response_callback(self, future): goal_handle future.result() if not goal_handle.accepted: self.get_logger().info(Goal rejected :() return self.get_logger().info(Goal accepted :)) # 目标被接受后异步请求最终结果 self._get_result_future goal_handle.get_result_async() # 绑定最终结果的回调 self._get_result_future.add_done_callback(self.get_result_callback) # 回调2 最终结果的回调 def get_result_callback(self, future): # 任务完成获取并打印最终结果 result future.result().result self.get_logger().info(fFinal Result: {result.result}) # 回调3 实时反馈的回调 def feedback_callback(self, feedback_msg): # 接收到服务器发来的实时反馈 feedback feedback_msg.feedback self.get_logger().info(fReceived feedback: {feedback.current}) sshi def main(): # 1 初始化环境 rclpy.init() # 2 创建节点 action_client_node ActionClientNode() # 2.1 发送信息 action_client_node.send_goal(10.0) # 3 消息循环 rclpy.spin(action_client_node) # 4 释放资源 action_client_node.destroy_node() rclpy.shutdown() if __name__ __main__: main()3 创建服务端import rclpy from rclpy.node import Node from rclpy.action import ActionServer from rclpy.action.server import ServerGoalHandle, GoalResponse from pkg_interfaces.action import MyAction class ActionServerNode(Node): def __init__(self): super().__init__(action_server) # 实例化动作服务器绑定动作类型、动作名称和执行回调函数 self._action_server ActionServer( self, MyAction, MyAction, self.execute_callback,# 执行回调用于反馈实时执行结果 feedback。当执行完成后反馈执行result消息 goal_callbackself.goal_callback)# 用于确定是否提供服务的回调如果不设置则默认回调提供服务 # 步骤1 是否提供服务回调 def goal_callback(self,goal_request): if goal_request.aim 0: self.get_logger().info(GoalResponse.REJECT) return GoalResponse.REJECT else: self.get_logger().info(GoalResponse.ACCEPT) return GoalResponse.ACCEPT # 步骤2 执行状态回调 def execute_callback(self, goal_handle: ServerGoalHandle): self.get_logger().info(Executing goal...) # 1. 初始化反馈消息 feedback_msg MyAction.Feedback() # 2. 获取客户端传来的目标值 aim goal_handle.request.aim # 3. 执行长时间运行的任务并发布反馈结果 for i in range(0, int(aim)): # 计算下一个数 feedback_msg.current float(i) # 发布实时反馈 goal_handle.publish_feedback(feedback_msg) self.get_logger().info(fFeedback: {feedback_msg.current}) # 模拟耗时操作实际工程中应使用定时器或异步等待 import time time.sleep(1) # 4. 任务完成设置成功状态 goal_handle.succeed() # 5. 返回result结果 result MyAction.Result() result.result True self.get_logger().info(Goal succeeded!) return result def main(): # 1 初始化资源 rclpy.init() # 2 创建节点 action_server_node ActionServerNode() action_server_node.get_logger().info(server started) # 3 消息循环 rclpy.spin(action_server_node) # 4 释放资源 action_server_node.destroy_node() rclpy.shutdown() if __name__ __main__: main()4 运行在setup.py中添加入口节点action_client_node pkg_name1.action_client_node:main, action_server_node pkg_name1.action_server_node:main,运行结果