ROS2中的Action机制—构建复杂机器人系统的关键工具
·
一、Action的核心概念
什么是Action？
Action是ROS 2中用于长时间运行任务的通信机制，它结合了Topic的实时反馈和Service的请求-响应模式，专门解决以下场景：
| 场景 | 为什么不用Service/Topic |
|---|---|
| 机器人导航到目标点 | Service会阻塞太久，Topic无法确认完成 |
| 机械臂执行复杂轨迹 | 需要中途取消或获取进度反馈 |
| 长时间图像处理 | 客户端需要知道处理进度和最终结果 |
Action vs Service vs Topic
┌─────────────────────────────────────────────────────────────┐
│ Topic (单向流) Service (同步调用) Action (异步任务) │
├─────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │Publisher │───────────▶│ Client │───────────▶│ Client │ │
│ └──────────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ │ │ │ │
│ ┌──────────┐ ┌────┴─────┐ ┌────┴─────┐ │
│ │Subscriber│◀───────────│ Server │ │ Server │ │
│ └──────────┘ └──────────┘ └────┬─────┘ │
│ ┌───────┼───────┐│
│ │Feedback│Result││
│ └───────┴───────┘│
└─────────────────────────────────────────────────────────────┘
二、Action的架构组成
2.1 三条通信通道
一个Action实际上由三个Topic和一个Service组成：
┌────────────────────────────────────────┐
│ Action: Fibonacci │
├────────────────────────────────────────┤
│ 1. Goal Service (/_action/send_goal) │
│ └─ 客户端发送目标，服务端确认接受 │
│ │
│ 2. Cancel Service (/_action/cancel_goal)│
│ └─ 客户端请求取消任务 │
│ │
│ 3. Status Topic (/_action/status) │
│ └─ 发布所有活跃目标的状态(心跳) │
│ │
│ 4. Feedback Topic (/_action/feedback) │
│ └─ 服务端发布执行进度(流式数据) │
│ │
│ 5. Result Service (/_action/get_result)│
│ └─ 客户端获取最终结果 │
└────────────────────────────────────────┘
2.2 数据类型定义 (.action 文件)
# Request (Goal)
int32 order
---
# Response (Result)
int32[] sequence
---
# Feedback (进度更新)
int32[] partial_sequence
注意：三个部分用
---分隔，分别对应 Goal、Result、Feedback
三、完整代码实战
3.1 定义Action接口
创建 action_tutorials_interfaces/action/Fibonacci.action:
int32 order
---
int32[] sequence
---
int32[] partial_sequence
在 CMakeLists.txt 中添加：
find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
"action/Fibonacci.action"
)
3.2 服务端 (Action Server)
# fibonacci_action_server.py
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from action_tutorials_interfaces.action import Fibonacci
import time
class FibonacciActionServer(Node):
def __init__(self):
super().__init__('fibonacci_action_server')
# 创建Action Server
self._action_server = ActionServer(
self,
Fibonacci,
'fibonacci',
self.execute_callback,
# 可选回调：
goal_callback=self.goal_callback, # 接受/拒绝目标
cancel_callback=self.cancel_callback, # 处理取消请求
handle_accepted_callback=self.handle_accepted_callback # 目标被接受后
)
self.get_logger().info('Action Server已启动')
def goal_callback(self, goal_request):
"""决定接受或拒绝目标"""
self.get_logger().info(f'收到目标: order={goal_request.order}')
if goal_request.order > 20:
self.get_logger().warn('拒绝: order太大(>20)')
return rclpy.action.GoalResponse.REJECT
self.get_logger().info('接受目标')
return rclpy.action.GoalResponse.ACCEPT
def cancel_callback(self, goal_handle):
"""处理取消请求"""
self.get_logger().info('收到取消请求')
return rclpy.action.CancelResponse.ACCEPT
def handle_accepted_callback(self, goal_handle):
"""目标被接受后的处理（如排队、立即执行等）"""
# 这里可以检查是否已有正在执行的目标
# 默认行为：直接执行
goal_handle.execute()
async def execute_callback(self, goal_handle):
"""实际执行逻辑（在单独线程中运行）"""
self.get_logger().info('开始执行...')
goal = goal_handle.request.order
feedback_msg = Fibonacci.Feedback()
feedback_msg.partial_sequence = [0, 1]
for i in range(1, goal):
# 检查是否收到取消请求
if goal_handle.is_cancel_requested:
goal_handle.canceled()
self.get_logger().info('任务已取消')
return Fibonacci.Result()
# 计算斐波那契数列
feedback_msg.partial_sequence.append(
feedback_msg.partial_sequence[i] + feedback_msg.partial_sequence[i-1]
)
# 发布反馈
self.get_logger().info(f'反馈: {feedback_msg.partial_sequence}')
goal_handle.publish_feedback(feedback_msg)
# 模拟耗时操作
time.sleep(1)
# 设置最终结果
goal_handle.succeed()
result = Fibonacci.Result()
result.sequence = feedback_msg.partial_sequence
self.get_logger().info(f'完成: {result.sequence}')
return result
def main(args=None):
rclpy.init(args=args)
server = FibonacciActionServer()
rclpy.spin(server)
server.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
3.3 客户端 (Action Client)
# fibonacci_action_client.py
import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from action_tutorials_interfaces.action import Fibonacci
class FibonacciActionClient(Node):
def __init__(self):
super().__init__('fibonacci_action_client')
self._action_client = ActionClient(self, Fibonacci, 'fibonacci')
def send_goal(self, order):
"""发送目标并设置回调"""
goal_msg = Fibonacci.Goal()
goal_msg.order = order
# 等待服务端上线
self.get_logger().info('等待Action Server...')
self._action_client.wait_for_server()
# 发送目标（异步）
self.get_logger().info(f'发送目标: 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 feedback_callback(self, feedback_msg):
"""接收进度反馈（实时）"""
feedback = feedback_msg.feedback
self.get_logger().info(
f'收到反馈: {feedback.partial_sequence}'
)
def get_result_callback(self, future):
"""接收最终结果"""
result = future.result().result
status = future.result().status
if status == rclpy.action.GoalStatus.STATUS_SUCCEEDED:
self.get_logger().info(f'成功! 结果: {result.sequence}')
else:
self.get_logger().warn(f'任务失败，状态码: {status}')
# 关闭节点
rclpy.shutdown()
def cancel_goal(self, goal_handle):
"""取消正在执行的目标"""
self.get_logger().info('请求取消...')
cancel_future = goal_handle.cancel_goal_async()
cancel_future.add_done_callback(self.cancel_done)
def cancel_done(self, future):
"""处理取消响应"""
cancel_response = future.result()
if len(cancel_response.goals_canceling) > 0:
self.get_logger().info('取消成功')
else:
self.get_logger().info('取消失败或已完成')
def main(args=None):
rclpy.init(args=args)
client = FibonacciActionClient()
client.send_goal(10)
rclpy.spin(client)
if __name__ == '__main__':
main()
四、工作流程时序图
客户端 服务端
│ │
│────── 1. send_goal() ─────────▶│
│ │
│◀───── 2. GoalResponse ────────│ (接受/拒绝)
│ │
│ │────── 3. 开始执行 ────▶
│ │ (execute_callback)
│ │
│◀───── 4. Feedback (流式) ─────│ (publish_feedback)
│◀──────────────────────────────│
│◀──────────────────────────────│
│ │
│ │────── 5. 完成 ────────▶
│ │ (succeed()/abort())
│ │
│────── 6. get_result() ────────▶│
│ │
│◀───── 7. Result ──────────────│
│ │
[可选] 中途取消:
│────── X. cancel_goal() ───────▶│
│ │
│◀───── X. 中断执行 ────────────│
五、关键特性深度解析
5.1 目标ID与生命周期
每个目标都有唯一UUID，服务端可同时管理多个目标：
# 获取目标ID
goal_id = goal_handle.goal_id
print(f'目标ID: {goal_id.uuid}')
# 查询目标状态
status = goal_handle.status # EXECUTING, SUCCEEDED, CANCELED等
5.2 并发控制
默认情况下服务端串行执行目标，可通过配置实现并发：
ActionServer(
self,
Fibonacci,
'fibonacci',
self.execute_callback,
handle_accepted_callback=self.handle_accepted_callback
)
def handle_accepted_callback(self, goal_handle):
"""自定义调度策略"""
# 策略1: 立即执行（默认）
goal_handle.execute()
# 策略2: 存入队列，按序执行
# self.goal_queue.put(goal_handle)
# 策略3: 拒绝新目标（如果已有执行中）
# if self.current_goal:
# goal_handle.abort()
5.3 可靠性保证
| 机制 | 说明 |
|---|---|
| Status Topic | 2Hz心跳，客户端监控服务端存活 |
| Goal Expiration | 目标可设置超时时间 |
| Result Caching | 服务端缓存结果，支持客户端重连获取 |
六、调试与工具
# 查看Action列表
ros2 action list
# 查看Action类型
ros2 action info /fibonacci
# 发送测试目标（命令行）
ros2 action send_goal /fibonacci action_tutorials_interfaces/action/Fibonacci "{order: 5}"
# 查看实时反馈（添加--feedback）
ros2 action send_goal /fibonacci action_tutorials_interfaces/action/Fibonacci "{order: 5}" --feedback
# 取消目标
ros2 action cancel_goal <goal-id>
七、最佳实践
- 耗时操作必须检查取消：在
execute_callback中定期调用goal_handle.is_cancel_requested - 反馈频率控制：避免过高频率发布feedback（建议10Hz以下）
- 异常处理：使用
try-except包裹执行逻辑，确保异常时调用goal_handle.abort() - 线程安全：
execute_callback在单独线程运行，注意数据竞争
Action是ROS 2中最复杂的通信原语，但掌握后能实现鲁棒的长时间任务管理，是构建复杂机器人系统的关键工具。
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
5
0- 0
已为社区贡献28条内容
所有评论(0)