ROS2--动作通信

ROS2动作通信详解

1️⃣ 核心概念

动作（Action）是ROS2中用于长时间运行任务的通信机制，包含三个核心部分：

• Goal：客户端发送的任务目标（如"移动到坐标(5,3)"）
• Feedback：执行过程中的进度反馈（如"当前已移动3米"）
• Result：任务完成后的最终结果（如"成功到达目标"）

2️⃣ 与服务/话题的区别

特性	服务(Service)	话题(Topic)	动作(Action)
通信模式	同步请求-响应	异步发布-订阅	异步目标-反馈-结果
适用场景	快速完成的简单操作	持续数据流	长时间运行、需要进度反馈的任务
典型应用	开关传感器	传感器数据流	导航任务、机械臂控制
是否支持取消	❌	❌	✔️

3️⃣ 动作文件定义

创建.action文件定义通信结构（示例：MoveRobot.action）：

# Goal定义
geometry_msgs/Point target
---
# Result定义
bool success
string message
---
# Feedback定义
float32 progress
string status

4️⃣ 通信流程

5️⃣ 服务端代码示例

import rclpy
from action_msgs.msg import GoalStatus
from rclpy.action import ActionServer

class MoveRobotServer(Node):
    def __init__(self):
        super().__init__('move_robot_server')
        self._action_server = ActionServer(
            self,
            MoveRobot,
            'move_robot',
            self.execute_callback)
    
    async def execute_callback(self, goal_handle):
        # 执行任务
        while not done:
            # 发送反馈
            feedback_msg = MoveRobot.Feedback()
            feedback_msg.progress = current_progress
            goal_handle.publish_feedback(feedback_msg)
            
        # 返回结果
        goal_handle.succeed()
        result = MoveRobot.Result()
        result.success = True
        return result

6️⃣ 客户端代码示例

from rclpy.action import ActionClient

class MoveRobotClient(Node):
    def __init__(self):
        super().__init__('move_robot_client')
        self._action_client = ActionClient(self, MoveRobot, 'move_robot')
    
    def send_goal(self, target):
        goal_msg = MoveRobot.Goal()
        goal_msg.target = target
        
        self._action_client.wait_for_server()
        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 feedback_callback(self, feedback_msg):
        print(f'当前进度: {feedback_msg.progress}%')
    
    def goal_response_callback(self, future):
        goal_handle = future.result()
        if not goal_handle.accepted:
            return
            
        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
        print(f'任务完成! 结果: {result.message}')

7️⃣ 常用命令行工具

# 查看可用动作列表
ros2 action list

# 查看动作类型
ros2 action info /move_robot

# 手动发送目标
ros2 action send_goal /move_robot MoveRobot "{target: {x: 5.0, y: 3.0}}"

8️⃣ 适用场景建议

✅ 使用动作的时机：

• 需要超过1秒的长时间任务
• 需要持续进度反馈
• 需要支持任务取消
• 需要处理可能的执行失败

❌ 不使用动作的情况：

• 瞬时完成的简单请求（改用服务）
• 单向数据流（改用话题）
• 需要实时性极高的通信（考虑自定义消息类型）

---

生命中真正重要的不是你遭遇了什么，而是你记住了哪些事，又是如何铭记的。 —加西亚·马尔克斯-\