当前位置： 首页 > news >正文

ROS2 Humble下用Python写Action服务端与客户端：一个模拟机器人移动的完整示例

news 2026/5/1 1:29:29

ROS2 Humble下Python Action开发实战：从机器人状态机到多线程优化

在机器人开发中，异步任务处理是个永恒的话题。想象一下，当你需要让机器人移动2米的同时还要实时监测环境变化，或者在进行机械臂轨迹规划时允许用户随时取消当前操作——这正是ROS2 Action设计的初衷。与简单的服务调用不同，Action提供了三大核心优势：长时间任务执行、实时进度反馈和任务取消机制。本文将带你用Python从零构建一个完整的机器人移动模拟系统，深入剖析rclpy.action的设计哲学，并解决实际开发中那些官方文档没告诉你的"坑"。

1. 环境准备与工程架构

1.1 创建Python功能包

在ROS2 Humble环境下，我们首先需要建立一个标准的Python功能包。与C++不同，Python包的创建有一些特殊配置需要注意：

ros2 pkg create robot_action_py \ --build-type ament_python \ --dependencies rclpy robot_control_interfaces \ --node-name action_server action_client \ --maintainer-name "your_name" \ --maintainer-email "your_email@domain.com"

关键参数解析：

  • --build-type ament_python：指定Python包类型
  • --dependencies：声明依赖的ROS2接口
  • --node-name：自动生成节点入口文件

提示：ROS2 Python包的目录结构遵循Python模块规范，所有源代码应放在<package_name>/<package_name>目录下

1.2 工程文件结构规划

一个规范的Action项目通常包含以下文件：

robot_action_py/ ├── robot_action_py/ │ ├── __init__.py │ ├── robot.py # 机器人模拟类 │ ├── action_server.py # Action服务端 │ └── action_client.py # Action客户端 ├── setup.py └── package.xml

在setup.py中需要确保正确配置入口点：

entry_points={ 'console_scripts': [ 'action_server = robot_action_py.action_server:main', 'action_client = robot_action_py.action_client:main', ], },

2. 机器人状态机实现

2.1 设计移动控制状态机

在robot.py中，我们实现一个具有完整状态管理的机器人类：

from robot_control_interfaces.action import MoveRobot import math class Robot: """模拟二维平面移动机器人""" def __init__(self): self._current_pose = 0.0 # 当前位置 self._target_pose = 0.0 # 目标位置 self._status = MoveRobot.Feedback.STATUS_READY self._move_step = 0.1 # 单步移动距离系数 def set_goal(self, distance): """设置移动目标""" if self._status != MoveRobot.Feedback.STATUS_READY: return False self._target_pose = self._current_pose + distance self._status = MoveRobot.Feedback.STATUS_MOVING return True def move_step(self): """执行单步移动""" if self._status != MoveRobot.Feedback.STATUS_MOVING: return False direction = math.copysign(1, self._target_pose - self._current_pose) step_size = direction * min( abs(self._target_pose - self._current_pose), self._move_step ) self._current_pose += step_size if self._check_goal_reached(): self._status = MoveRobot.Feedback.STATUS_COMPLETED return True def _check_goal_reached(self): """检查是否到达目标""" return math.isclose( self._current_pose, self._target_pose, abs_tol=0.01 ) def get_feedback(self): """生成反馈消息""" feedback = MoveRobot.Feedback() feedback.pose = self._current_pose feedback.status = self._status return feedback

状态转换图示意：

[READY] -> [MOVING] -> [COMPLETED] ↑ | └----------┘ (通过set_goal重置)

2.2 异常处理机制

完善的机器人控制需要考虑各种异常情况：

class Robot: # ...原有代码... def emergency_stop(self): """紧急停止""" self._status = MoveRobot.Feedback.STATUS_EMERGENCY_STOP def handle_cancel(self): """处理取消请求""" if self._status == MoveRobot.Feedback.STATUS_MOVING: self._status = MoveRobot.Feedback.STATUS_CANCELLED return True return False

3. Action服务端深度解析

3.1 服务端核心架构

在action_server.py中构建完整的Action服务：

import rclpy from rclpy.action import ActionServer from rclpy.node import Node from robot_control_interfaces.action import MoveRobot from .robot import Robot class RobotActionServer(Node): def __init__(self): super().__init__('robot_action_server') self.robot = Robot() self._action_server = ActionServer( self, MoveRobot, 'move_robot', self._execute_callback, goal_callback=self._goal_callback, cancel_callback=self._cancel_callback ) self.get_logger().info("Action服务器已启动...") def _goal_callback(self, goal_request): """处理新目标请求""" self.get_logger().info(f'收到新目标: 移动 {goal_request.distance}米') if not self.robot.set_goal(goal_request.distance): return rclpy.action.server.GoalResponse.REJECT return rclpy.action.server.GoalResponse.ACCEPT def _cancel_callback(self, cancel_request): """处理取消请求""" self.get_logger().info('收到取消请求') if self.robot.handle_cancel(): return rclpy.action.server.CancelResponse.ACCEPT return rclpy.action.server.CancelResponse.REJECT async def _execute_callback(self, goal_handle): """执行移动任务""" self.get_logger().info('开始执行移动任务...') while rclpy.ok() and not goal_handle.is_cancel_requested: if not self.robot.move_step(): break feedback = self.robot.get_feedback() goal_handle.publish_feedback(feedback) await asyncio.sleep(0.1) # 非阻塞式等待 result = MoveRobot.Result() result.pose = self.robot.get_feedback().pose if goal_handle.is_cancel_requested: goal_handle.canceled() self.get_logger().warn('任务被取消') elif self.robot.get_feedback().status == MoveRobot.Feedback.STATUS_COMPLETED: goal_handle.succeed() self.get_logger().info('任务完成!') return result

关键设计要点：

  • 使用异步execute_callback避免阻塞主线程
  • 明确的状态反馈机制
  • 完善的取消处理流程

3.2 多线程执行器优化

默认的单线程执行器会导致rate.sleep()死锁问题，解决方案：

from rclpy.executors import MultiThreadedExecutor def main(args=None): rclpy.init(args=args) server = RobotActionServer() # 使用多线程执行器 executor = MultiThreadedExecutor(num_threads=4) executor.add_node(server) try: executor.spin() except KeyboardInterrupt: server.get_logger().info('服务器关闭中...') finally: server.destroy_node() rclpy.shutdown()

线程分配策略：

  • 1个线程处理Action通信
  • 1个线程执行任务处理
  • 2个备用线程处理其他回调

4. 智能Action客户端实现

4.1 客户端状态管理

在action_client.py中实现带超时管理的客户端：

import rclpy from rclpy.action import ActionClient from rclpy.node import Node from robot_control_interfaces.action import MoveRobot class RobotActionClient(Node): def __init__(self): super().__init__('robot_action_client') self._client = ActionClient(self, MoveRobot, 'move_robot') self._goal_handle = None self._send_goal_timer = self.create_timer(1.0, self.send_goal) def send_goal(self): """发送移动目标""" self._send_goal_timer.cancel() goal_msg = MoveRobot.Goal() goal_msg.distance = 2.5 # 默认移动2.5米 self._client.wait_for_server(timeout_sec=5.0) self._send_goal_future = self._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): """处理目标响应""" self._goal_handle = future.result() if not self._goal_handle.accepted: self.get_logger().error('目标被拒绝') return self.get_logger().info('目标已接受，执行中...') self._get_result_future = self._goal_handle.get_result_async() self._get_result_future.add_done_callback(self._result_callback) def _feedback_callback(self, feedback_msg): """处理进度反馈""" feedback = feedback_msg.feedback self.get_logger().info( f'当前进度: {feedback.pose:.2f}m | 状态: {feedback.status}' ) def _result_callback(self, future): """处理最终结果""" result = future.result().result self.get_logger().info(f'最终位置: {result.pose:.2f}m') rclpy.shutdown()

4.2 高级功能扩展

为客户端添加更多实用功能：

class RobotActionClient(Node): # ...原有代码... def cancel_goal(self): """取消当前目标""" if self._goal_handle is not None: future = self._goal_handle.cancel_goal_async() 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().warning('取消请求被拒绝')

5. 调试技巧与性能优化

5.1 常见问题排查指南

问题现象可能原因解决方案
Action调用无响应服务端未启动检查ros2 action list输出
反馈信息延迟单线程阻塞使用MultiThreadedExecutor
取消请求无效未正确处理回调实现cancel_callback
目标频繁被拒状态机未重置检查机器人ready状态

5.2 性能优化策略

  1. 回调分组优化：
from rclpy.callback_groups import ReentrantCallbackGroup self._action_server = ActionServer( self, MoveRobot, 'move_robot', execute_callback=self._execute_callback, callback_group=ReentrantCallbackGroup() )
  1. QoS配置调整：
from rclpy.qos import QoSProfile action_qos = QoSProfile( depth=10, reliability=rclpy.qos.ReliabilityPolicy.RELIABLE ) self._action_server = ActionServer( # ...其他参数... feedback_pub_qos_profile=action_qos )
  1. 执行器线程数建议：
  • 简单场景：2-4个线程
  • 复杂系统：CPU核心数×1.5

在实际机器人项目中，Action通信的稳定性往往决定了整个系统的可靠性。记得在正式部署前进行压力测试，模拟网络延迟和节点重启等情况。我曾在一个仓储机器人项目中发现，当Action消息频率超过50Hz时，需要特别优化反馈消息的数据量，否则会导致整个系统延迟增加。

http://www.jsqmd.com/news/728678/

相关文章：

  • 手把手教你用另一个JLink救活变砖的JLink V9（附接线图与固件下载）
  • 从 0 到 1 落地 AI 客服：基于冰石智能平台的提示词实战与避坑指南
  • ARM浮点运算指令FMLS与FMSUB详解与应用优化
  • 终极游戏模组管理器：XXMI启动器让你一键管理所有二次元游戏模组
  • 别再只会用gdb了！用objdump反编译Linux程序，5分钟看懂别人代码逻辑
  • 9、OpenClaw（龙虾助手）哔哩哔哩完整对接指南（2026最新版）
  • 机器学习大师课 第 4 课：分类问题入门 —— 逻辑回归（垃圾邮件分类实战）
  • Java异步编程与资源管理笔记
  • 告别默认‘滴滴’声！用Bluejay Configurator给你的穿越机电调定制专属开机BGM（附天空之城、JOJO等曲谱）
  • Pine64 StarPro64 RISC-V开发板：高性能与AI加速解析
  • 使用Taotoken后如何清晰查看API用量与成本分布
  • Day1 C与python输入输出语句区别
  • 魔兽争霸3帧率优化指南：如何通过开源工具WarcraftHelper突破60帧限制
  • VCS后仿保姆级避坑指南：从网表、SDF到lib库的完整配置流程
  • 思源宋体终极指南：7款免费商用字体快速上手与实战技巧
  • 2026年知网新算法下论文降AI收藏指南：降低AI率硬核手改技巧+降AI率工具实测 - 降AI实验室
  • CTP穿透式监管下，企业级量化系统如何设计订单与持仓管理模块？
  • 对话式数据可视化：用自然语言驱动Vega-Lite图表生成
  • 串口通信无线化方案与工业物联网应用
  • 超算小白避坑指南：用Slurm和Conda搞定深度学习环境（附常见错误排查）
  • 别让爬虫白嫖你的导航站了：纯免费，手把手实现加密字体防爬
  • 从零开始掌握LaserGRBL：开源激光雕刻软件的完整使用指南
  • 给你的STM32F103C8T6开发板“添砖加瓦”：ESP8266联网、OLED显示与蓝牙控制实战
  • 每个程序员都该学习的5种开发语言
  • 五指灵巧手有哪些技术要点？2026年专业五指灵巧手厂商甄选 - 品牌2026
  • 2026年深圳初升高品牌排行：聚焦民办高中核心竞争力 - 优质品牌商家
  • 2026年我的效率工具箱：5个每天必用的软件
  • 从账单明细看 Taotoken 计费透明性带来的管理便利
  • 2026铝皮保温玻璃棉品牌排行：铝皮保温保护层、铝皮保温加工、铝皮保温厂家、铝皮保温外壳、铝皮保温安装报价、铝皮保温施工费用选择指南 - 优质品牌商家
  • 如何彻底解决ComfyUI的GPU显存泄漏问题