4. ROS2动作通信基础
任务目标
本案例实现一个ROS2动作通信的最小闭环：动作服务器执行倒计时任务，客户端发送目标后，服务器周期性反馈剩余秒数，任务完成后返回最终结果。通过该案例掌握ROS2动作通信的三要素——目标、反馈、结果——以及客户端异步调用的基本流程。
环境依赖
- ROS2 Humble或更高版本（推荐Humble）
- 已安装
rclpy和action_tutorials_interfaces（本案例使用该包中的Fibonacci动作类型，无需额外定义）
实现代码
将以下代码保存为countdown_action_server.py和countdown_action_client.py，放在同一目录下。
# countdown_action_server.py
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from action_tutorials_interfaces.action import Fibonacci
class CountdownServer(Node):
def __init__(self):
super().__init__('countdown_server')
self._action_server = ActionServer(
self,
Fibonacci,
'countdown',
execute_callback=self.execute_callback)
async def execute_callback(self, goal_handle):
self.get_logger().info(f'收到目标: {goal_handle.request.order} 秒倒计时')
feedback_msg = Fibonacci.Feedback()
remaining = goal_handle.request.order
while remaining > 0:
feedback_msg.sequence = [remaining]
goal_handle.publish_feedback(feedback_msg)
self.get_logger().info(f'剩余: {remaining} 秒')
await asyncio.sleep(1.0)
remaining -= 1
goal_handle.succeed()
result = Fibonacci.Result()
result.sequence = [0]
self.get_logger().info('倒计时完成')
return result
def main(args=None):
rclpy.init(args=args)
import asyncio
node = CountdownServer()
executor = rclpy.executors.MultiThreadedExecutor()
executor.add_node(node)
executor.spin()
rclpy.shutdown()
if __name__ == '__main__':
main()
# countdown_action_client.py
import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from action_tutorials_interfaces.action import Fibonacci
class CountdownClient(Node):
def __init__(self):
super().__init__('countdown_client')
self._action_client = ActionClient(self, Fibonacci, 'countdown')
def send_goal(self, seconds):
goal_msg = Fibonacci.Goal()
goal_msg.order = seconds
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 goal_response_callback(self, future):
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().info('目标被拒绝')
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):
self.get_logger().info(f'收到反馈: 剩余 {feedback_msg.feedback.sequence[0]} 秒')
def get_result_callback(self, future):
result = future.result().result
self.get_logger().info(f'最终结果: {result.sequence[0]}')
rclpy.shutdown()
def main(args=None):
rclpy.init(args=args)
node = CountdownClient()
node.send_goal(5) # 请求5秒倒计时
rclpy.spin(node)
if __name__ == '__main__':
main()
运行与验证
- 安装依赖动作类型包：
sudo apt install ros-$ROS_DISTRO-action-tutorials-interfaces - 终端1启动服务器：
python3 countdown_action_server.py - 终端2启动客户端：
python3 countdown_action_client.py
通过条件：
- 服务器终端依次打印“收到目标: 5 秒倒计时”，随后每秒打印“剩余: 5/4/3/2/1 秒”，最后打印“倒计时完成”。
- 客户端终端打印“目标已接受”，随后每秒收到反馈“剩余 N 秒”（N从5递减到1），最后打印“最终结果: 0”。
- 两个终端均无报错，客户端在收到结果后自动退出。
关键说明
本案例复用Fibonacci.action类型（字段为order/sequence），将其语义映射为倒计时秒数。服务器使用asyncio.sleep模拟每秒递减，实际工程中可替换为真实任务进度查询。客户端采用异步回调方式接收反馈与结果，这是ROS2动作通信的标准模式。若需自定义动作类型，可参照ROS2官方教程创建.action文件并编译生成接口。