1. 从发完指令就失联说起动作通信到底补了哪块短板如果你跟着前面的内容一路学过来应该已经习惯了两种通信方式话题像广播电台发出去就不管了谁爱听谁听服务像打电话拨通了问一句答一句干脆利落。但真到做项目的时候你会发现这两种方式中间有一大片空白地带——那些要跑很久、中途还得汇报进度、跑完还得给个结果的任务用话题做吧没法确认对方到底收没收到、干没干完用服务做吧服务是同步阻塞的你这边一调用整个节点就卡在那儿等机器人要是正在跑导航主线程一卡整个系统就僵住了。动作通信就是专门来填这个坑的。它本质上是把话题和服务揉在一起用底层用一对话题来传目标和结果中间再挂一个话题专门播报进度同时支持任务执行到一半我不想干了取消掉这种操作。你可以把它理解成点外卖——你下单发目标骑手接单后你能看到已取餐正在配送中进度反馈最后送到你手上结果返回中途你还能点取消。这套机制在机器人领域几乎是刚需导航到某个点、机械臂抓取某个物体、无人机飞一条航线全都是典型的动作任务。这篇内容我打算把动作编程从能跑通讲到敢用在项目里。市面上很多教程只告诉你敲哪几行命令、看到什么输出就算成功但真到自己写动作服务端的时候进度怎么设计、取消怎么响应、超时怎么处理这些才是决定你能不能落地的关键。我会用一套自己反复调试过的代码结构把服务端、客户端、进度反馈、取消处理这几块拆开讲透中间穿插我踩过的坑和实测有效的参数设置。不管你是刚学完话题和服务想往上走一步还是已经在做项目被异步任务卡住了这篇应该都能给你一些能直接抄的东西。2. 动作通信的三段式结构目标、反馈、结果各管什么2.1 为什么不能只用服务同步阻塞的致命伤先把这个为什么讲清楚不然后面设计代码的时候容易想歪。服务调用的模型是请求-响应客户端发一个请求过去然后就在那儿等着直到服务端把响应发回来。这个等是阻塞的意味着你这行代码不执行完后面的代码一行都动不了。短任务无所谓几十毫秒的事但要是服务端要跑个三五秒甚至几十秒客户端节点基本就废了。有人会说那我开个新线程去调服务不就行了可以但问题是你得自己管理这个线程的生命周期还得自己想办法把进度传出来——而服务本身是没有进度这个概念的你只能再额外开一个话题专门发进度。等你把这些都手动拼起来其实你已经在重新发明动作通信了而且大概率没有官方实现得稳。动作通信就是把这套异步执行进度反馈可取消的模式标准化了你不用自己造轮子。2.2 三个核心接口Goal、Feedback、Result动作的接口定义里最关键的就是三个部分我习惯叫它们三段式Goal目标客户端发过去的任务描述。比如导航到坐标(x, y)、抓取编号为3的物体。这部分是请求性质的服务端要能判断自己接不接这个活。Feedback反馈服务端在执行过程中周期性发回来的进度信息。注意是周期性不是每时每刻频率要自己控制发太猛会把网络和CPU吃满。Result结果任务最终完成后返回的信息包含成功与否、最终数据等。这个只在任务真正结束时发一次。这三个部分在.action文件里定义格式上其实就是把三个.msg拼在一起用---隔开。我拿一个最经典的例子——数数动作——来演示这个例子简单到不会分散你对机制的注意力但该有的要素一个不少# 目标 int32 target_number --- # 结果 int32 final_count --- # 反馈 int32 current_number翻译成人话就是客户端说给我从1数到target_number服务端一边数一边把当前数到几了反馈回来数完了把最终结果返回。这个例子虽然简单但它把动作通信的三个通道全用上了非常适合拿来理解机制。2.3 底层其实是话题在跑别被动作这个名字唬住很多人第一次接触动作通信会觉得它是个全新的东西其实拆开看它底层就是几个话题在跑。目标、反馈、结果、取消每一个都对应一个话题只不过官方帮你把这些话题的命名规则、消息类型、状态机都封装好了。你写代码的时候用的是ActionClient和ActionServer这种高层接口但如果你用rostopic list去看是能看到底下那些话题的。理解这一点很重要因为它解释了为什么动作通信能天然支持异步——话题本来就是异步的。也解释了为什么动作的调试有时候要用到话题的工具比如你想看反馈到底发出来没有可以直接echo反馈话题比在代码里打日志还快。提示不同版本的ROS在动作通信的实现上有差异。ROS1用的是actionlib这套库ROS2里改成了rclcpp_action/rclpy_action接口风格变了但三段式的核心思想没变。这篇以ROS1的actionlib为主线讲因为它的概念最清晰学透了再迁移到ROS2会非常顺。3. 动手写一个动作服务端从定义到回调的完整链路3.1 先建功能包和action文件目录结构别搞错我见过太多人卡在第一步——action文件放错地方编译死活过不去。这里把目录结构说死你照着建就行。假设功能包叫action_tutorials那么action_tutorials/ ├── action/ │ └── CountTo.action ├── scripts/ │ ├── count_server.py │ └── count_client.py ├── CMakeLists.txt └── package.xml关键点在于action这个目录名不能随便改而且CMakeLists.txt里要显式声明这个action文件否则编译系统根本不知道要为你生成消息代码。在package.xml里要加上对actionlib和actionlib_msgs的依赖这两个是动作通信的命脉少一个都编译不过。CMakeLists.txt里最容易被漏掉的是这两行find_package(catkin REQUIRED COMPONENTS rospy std_msgs actionlib actionlib_msgs ) add_action_files( FILES CountTo.action ) generate_messages( DEPENDENCIES std_msgs actionlib_msgs )add_action_files负责把.action文件注册进去generate_messages负责生成对应的Python/C消息代码。这两步缺一不可而且顺序不能乱——generate_messages必须在add_action_files之后。我第一次搭的时候把顺序写反了编译报了一堆找不到消息类型的错排查了半天才发现是这个顺序问题。3.2 服务端的骨架SimpleActionServer还是ActionServer写服务端有两个选择SimpleActionServer和ActionServer。前者是后者的简化版它帮你处理了大部分状态机逻辑你只需要关心目标来了怎么办取消来了怎么办这几个回调。对于绝大多数项目SimpleActionServer完全够用我强烈建议新手从这里入手别一上来就啃ActionServer那套复杂的状态管理。SimpleActionServer的初始化长这样import rospy import actionlib from action_tutorials.msg import CountToAction, CountToFeedback, CountToResult class CountServer: def __init__(self): self.server actionlib.SimpleActionServer( count_to, CountToAction, execute_cbself.execute_cb, auto_startFalse ) self.server.start() rospy.loginfo(动作服务端已启动) def execute_cb(self, goal): # 任务执行逻辑写在这里 pass这里有个细节值得说auto_startFalse加上手动start()是为了确保服务端完全初始化好了再开始接收目标。如果你设成auto_startTrue理论上更省事但在复杂场景下可能出现服务端还没准备好就收到目标的情况所以我习惯手动控制启动时机。3.3 execute_cb里的进度与取消两个必须处理的信号execute_cb是整个服务端的核心它在一个独立的线程里跑所以你可以放心在里面写耗时的循环不会阻塞其他回调。但正因为它是独立线程有两个信号你必须主动去查否则会出问题取消信号客户端点了取消你得知道并且尽快停下来。查法是self.server.is_preempt_requested()返回True说明有人要取消。抢占信号新的目标来了旧目标要不要让位。这个和取消类似也是通过is_preempt_requested()判断。一个健壮的execute_cb应该长这样def execute_cb(self, goal): success True target goal.target_number rate rospy.Rate(1) # 1Hz每秒反馈一次 for i in range(1, target 1): # 每次循环先检查是否被取消 if self.server.is_preempt_requested(): rospy.loginfo(任务被取消) self.server.set_preempted() success False break # 发布进度反馈 feedback CountToFeedback() feedback.current_number i self.server.publish_feedback(feedback) rate.sleep() if success: result CountToResult() result.final_count target self.server.set_succeeded(result)这段代码里有几个我踩过坑的地方。第一is_preempt_requested()的检查必须放在循环里而且要在耗时操作之前。我见过有人把它放在循环外面结果取消信号根本不起作用客户端点了取消服务端还在傻跑。第二publish_feedback的频率要控制我用rospy.Rate(1)限制成1Hz实际项目里根据任务时长调整短任务可以高一点长任务1Hz足够了。第三set_preempted()和set_succeeded()是互斥的一个任务只能有一个最终状态别两个都调。注意execute_cb里千万不要写死循环而不检查取消信号。我见过一个导航任务的服务端循环里没有取消检查结果客户端取消后服务端还在跑最后两个目标叠在一起机器人直接懵了。这个坑的代价是撞了一次墙希望你别重蹈覆辙。4. 客户端怎么写才不假死回调、超时与结果获取4.1 发送目标wait_for_server不能省客户端的第一件事是等服务器上线。这一步很多人图省事直接跳过结果就是目标发出去石沉大海因为服务端还没起来。正确做法是先wait_for_server()client actionlib.SimpleActionClient(count_to, CountToAction) client.wait_for_server()wait_for_server()默认会一直等下去如果你不想无限等可以传个超时参数。在项目里我一般会设个5秒左右的超时超时就报错退出而不是傻等——因为服务端没起来通常意味着系统启动顺序有问题早点暴露比晚点暴露好。4.2 回调函数进度和结果分开处理客户端发完目标后不应该阻塞等待而是注册回调让进度和结果自动触发def feedback_cb(feedback): rospy.loginfo(当前进度%d % feedback.current_number) def done_cb(state, result): if state actionlib.GoalStatus.SUCCEEDED: rospy.loginfo(任务完成最终结果%d % result.final_count) else: rospy.loginfo(任务未成功完成状态码%d % state) client.send_goal(goal, done_cbdone_cb, feedback_cbfeedback_cb)这里done_cb的state参数很关键它告诉你任务到底是成功了、被取消了、还是被抢占了。我强烈建议在done_cb里对不同的state做不同处理而不是简单地打印完成。因为在实际项目里被取消和成功往往需要触发完全不同的后续逻辑。4.3 等待结果wait_for_result的阻塞陷阱如果你确实需要等结果可以用wait_for_result()但要知道它是阻塞的。在单线程的客户端里调用它之后你的程序就卡在那儿了直到任务结束。这在简单的测试脚本里没问题但在真正的机器人主控节点里阻塞主线程是危险的。我的做法是主控节点里永远用回调不用wait_for_result()。只有在写一次性测试脚本、或者明确知道这个客户端节点就是专门干这一件事的时候才用阻塞等待。这个习惯帮我避免了好几次机器人主控卡死的事故。# 测试脚本里可以这样写 client.send_goal(goal) client.wait_for_result() print(client.get_result())4.4 取消操作cancel_goal的时机与确认取消不是发出去就完事了你得确认服务端真的停了。cancel_goal()只是发出取消请求服务端什么时候响应、响应成什么样需要你在done_cb里通过state来判断。一个完整的取消流程是这样的client.cancel_goal() # 然后在done_cb里会收到state为PREEMPTED的回调我实测下来从发出取消到服务端真正停下来中间有个几十毫秒到几百毫秒的延迟取决于服务端循环里检查取消信号的频率。如果你的任务对取消响应时间有要求就要把检查频率调高比如把rospy.Rate(1)改成rospy.Rate(10)但代价是反馈发得更频繁要权衡。5. 调试动作通信时我踩过的那些坑5.1 反馈收不到八成是频率或话题名的问题反馈收不到是最常见的现象。排查顺序我总结成三步第一步用rostopic list看反馈话题在不在话题名一般是/count_to/feedback这种格式第二步用rostopic echo直接看话题上有没有数据如果有数据但客户端收不到那就是客户端回调注册的问题第三步如果话题上根本没数据那就是服务端publish_feedback没被调用回去检查execute_cb里的循环逻辑。我遇到过一次反馈时有时无的情况最后发现是rospy.Rate设得太高反馈发得太密网络缓冲区满了丢包。把频率从100Hz降到10Hz就稳了。所以反馈频率不是越高越好够用就行。5.2 目标被拒绝状态机的隐藏规则有时候客户端发目标服务端直接拒绝客户端收到REJECTED状态。这通常是因为服务端当前正在处理另一个目标而且没有开启抢占。SimpleActionServer默认是允许抢占的但如果你在初始化时设了auto_start相关的参数不当可能会出现拒绝的情况。排查这个问题的关键是看服务端的日志SimpleActionServer会打印状态转换的信息。如果看到Rejecting goal之类的字样基本就是状态机的问题。解决办法通常是检查服务端是不是卡在某个execute_cb里没出来导致新目标进不来。5.3 结果拿不到done_cb没触发done_cb不触发任务就像失联了一样。这种情况我遇到过两次一次是服务端execute_cb里抛了异常但没被捕获导致set_succeeded没执行任务永远处于进行中另一次是客户端在send_goal之后很快就退出了回调还没来得及执行。第一个问题的解决办法是在execute_cb里加try-except确保无论发生什么最终状态都会被设置def execute_cb(self, goal): try: # 任务逻辑 self.server.set_succeeded(result) except Exception as e: rospy.logerr(任务执行出错%s % str(e)) self.server.set_aborted()第二个问题就是客户端生命周期管理的问题确保客户端节点在任务完成前不要退出或者用wait_for_result()强制等待。5.4 多目标并发抢占还是排队当一个服务端同时收到多个目标时默认行为是抢占——新目标把旧目标踢掉。这在某些场景下是对的比如导航任务新目标来了当然要优先。但在另一些场景下是错的比如机械臂的抓取任务你不能抓一半被新任务打断。要改变这个行为需要在服务端做额外处理比如维护一个目标队列或者用ActionServer非Simple版自己管理状态机。这块比较复杂我的建议是如果你的项目需要排队先想清楚是不是真的需要很多时候用多个服务端实例或者在上层做调度反而更简单。6. 把动作通信用到真实项目里的几点经验6.1 进度设计别只报百分比很多人设计反馈的时候只报一个百分比比如完成了30%。这在简单任务里够用但在复杂任务里信息量太少。我习惯把反馈设计成结构化的包含当前阶段、已完成子任务数、预估剩余时间等多个字段。这样上层做可视化或者做决策的时候能拿到更多信息。比如一个巡检任务反馈里可以包含当前巡检到第几个点总共几个点当前点的检测结果。这些信息比一个干巴巴的百分比有用得多。6.2 超时处理客户端和服务端都要设超时要在两端都设。客户端设超时是为了防止服务端挂了之后自己无限等待服务端设超时是为了防止某个任务卡死导致整个服务端不可用。客户端的超时用wait_for_result(rospy.Duration(30))这种形式服务端的超时则要在execute_cb里自己判断任务跑了多久超了就主动set_aborted()。我一般会把服务端的超时设得比客户端的长一点比如客户端30秒服务端45秒。这样正常情况下是客户端先超时并取消服务端收到取消后正常退出两边都能干净收场。6.3 日志与可观测性动作调试的命根子动作通信因为涉及异步和多状态出问题的时候光看现象很难定位。我的经验是在服务端和客户端的关键节点都打日志目标收到时打一条、开始执行时打一条、每次反馈时打一条频率高的话可以降采样、状态转换时打一条、结束时打一条。这些日志在排查问题的时候能帮你快速还原整个时间线。rospy.loginfo配合rqt_console用起来很顺手可以按节点过滤、按级别过滤。我调试动作通信的时候rqt_console基本是常开的。6.4 从ROS1迁移到ROS2概念不变接口重学如果你后面要转到ROS2动作通信的核心概念——目标、反馈、结果、取消——是完全一样的但接口全变了。ROS2里用rclcpp_action或rclpy.action客户端叫ActionClient服务端叫ActionServer回调的注册方式、结果获取方式都不一样。我的建议是先把ROS1这套吃透理解了三段式和状态机再去看ROS2的文档会发现只是换了个写法思想没变。迁移的时候最容易踩的坑是ROS2的动作接口对生命周期管理更严格节点的创建、执行器的选择都会影响动作能不能正常工作。这块等你有ROS2基础了再深入现在先把ROS1这套跑通。7. 一套可以直接复用的代码模板7.1 服务端模板带取消检查和异常保护把前面讲的东西整合成一个可以直接抄的模板#!/usr/bin/env python import rospy import actionlib from action_tutorials.msg import CountToAction, CountToFeedback, CountToResult class CountServer: def __init__(self): self.server actionlib.SimpleActionServer( count_to, CountToAction, execute_cbself.execute_cb, auto_startFalse) self.server.start() rospy.loginfo(CountServer started) def execute_cb(self, goal): try: target goal.target_number rate rospy.Rate(10) for i in range(1, target 1): if self.server.is_preempt_requested(): rospy.loginfo(Preempted) self.server.set_preempted() return fb CountToFeedback() fb.current_number i self.server.publish_feedback(fb) rate.sleep() result CountToResult() result.final_count target self.server.set_succeeded(result) except Exception as e: rospy.logerr(Error: %s % str(e)) self.server.set_aborted() if __name__ __main__: rospy.init_node(count_server) server CountServer() rospy.spin()这个模板里取消检查、异常保护、反馈发布、结果设置都齐了改改业务逻辑就能用。7.2 客户端模板回调驱动不阻塞#!/usr/bin/env python import rospy import actionlib from action_tutorials.msg import CountToAction, CountToGoal def feedback_cb(fb): rospy.loginfo(Feedback: %d % fb.current_number) def done_cb(state, result): if state actionlib.GoalStatus.SUCCEEDED: rospy.loginfo(Done: %d % result.final_count) else: rospy.loginfo(Failed with state: %d % state) if __name__ __main__: rospy.init_node(count_client) client actionlib.SimpleActionClient(count_to, CountToAction) client.wait_for_server() goal CountToGoal() goal.target_number 10 client.send_goal(goal, done_cbdone_cb, feedback_cbfeedback_cb) rospy.spin()注意最后用rospy.spin()而不是wait_for_result()这样客户端不会阻塞回调能正常触发。7.3 测试流程从命令行到代码跑通这套代码的完整流程是先rosrun启动服务端再rosrun启动客户端观察客户端的日志输出。如果反馈和结果都正常打印说明链路通了。然后可以试试在客户端运行过程中按CtrlC看服务端能不能正确响应取消。这一步很多人不做但恰恰是验证取消逻辑的关键。命令行工具rostopic echo /count_to/feedback可以单独看反馈话题rostopic echo /count_to/result看结果话题rostopic echo /count_to/status看状态话题。这三个话题是动作通信的仪表盘调试的时候非常有用。8. 写在最后的一点个人体会动作通信是我在机器人项目里用得最多的通信方式之一它的价值不在于多复杂而在于它把异步任务这个在工程里反复出现的问题标准化了。你学会了这一套以后遇到任何要跑一会儿、要报进度、要能取消的需求脑子里第一个跳出来的就是它。我自己的经验是动作通信的难点从来不在语法而在设计。进度怎么设计才有用、取消响应多快才够、超时设多少才合理这些没有标准答案得根据你的具体场景去调。我建议你在学完基础之后找一个自己项目里的真实任务用动作通信重写一遍把进度、取消、超时都加上跑起来看看哪里别扭。这个过程比看十篇教程都管用。最后分享一个小技巧如果你不确定某个动作任务该不该用动作通信就问自己三个问题——这个任务要跑超过一秒吗需要知道进度吗需要中途取消吗三个里有两个是是那就用动作通信别犹豫。