4. ROS2动作通信基础
任务目标
本案例实现一个ROS2动作通信的最小闭环：动作服务器执行倒计时任务，客户端发送目标请求并接收进度反馈与最终结果，同时使用ROS2自带的action命令行工具查看动作状态。通过本案例掌握动作通信的三要素——目标(Goal)、反馈(Feedback)、结果(Result)——及其异步交互模型。
环境与依赖
- 已安装ROS2（Humble或更高版本），并完成
source /opt/ros/<distro>/setup.bash - 创建工作空间并编译：
mkdir -p ~/ros2_ws/src && cd ~/ros2_ws/src && ros2 pkg create action_tutorials --build-type ament_python --dependencies rclpy rclpy.action action_msgs
实现步骤
动作接口使用ROS2内置的example_interfaces/action/Fibonacci，该接口包含goal（目标阶数）、feedback（部分序列）和result（完整序列）。本案例将其复用为倒计时任务：目标值表示倒计时起始秒数，反馈每秒输出剩余秒数，结果返回倒计时总耗时。
关键完整代码
在action_tutorials/action_tutorials/目录下创建countdown_server.py：
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from example_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()
result = Fibonacci.Result()
for remaining in range(goal_handle.request.order, 0, -1):
feedback_msg.sequence = [remaining]
self.get_logger().info(f'剩余: {remaining} 秒')
goal_handle.publish_feedback(feedback_msg)
await rclpy.sleep(1.0)
goal_handle.succeed()
result.sequence = [0]
self.get_logger().info('倒计时完成')
return result
def main(args=None):
rclpy.init(args=args)
node = CountdownServer()
try:
rclpy.spin(node)
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
创建countdown_client.py：
import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from example_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 feedback_callback(self, feedback_msg):
remaining = feedback_msg.feedback.sequence[0]
self.get_logger().info(f'客户端收到反馈: 剩余 {remaining} 秒')
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.result_callback)
def result_callback(self, future):
result = future.result().result
self.get_logger().info(f'客户端收到结果: 倒计时结束，序列={result.sequence}')
rclpy.shutdown()
def main(args=None):
rclpy.init(args=args)
node = CountdownClient()
try:
node.send_goal(5)
rclpy.spin(node)
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
在setup.py中注册入口点：
entry_points={
'console_scripts': [
'countdown_server = action_tutorials.countdown_server:main',
'countdown_client = action_tutorials.countdown_client:main',
],
},
运行与验证
- 编译并source：
cd ~/ros2_ws && colcon build --packages-select action_tutorials && source install/setup.bash - 终端1运行服务器：
ros2 run action_tutorials countdown_server - 终端2运行客户端：
ros2 run action_tutorials countdown_client - 终端3查看动作状态：
ros2 action list -t应显示/countdown [example_interfaces/action/Fibonacci] - 终端3发送目标并观察反馈：
ros2 action send_goal /countdown example_interfaces/action/Fibonacci "{order: 3}" --feedback
通过条件：
- 服务器终端按秒递减打印剩余时间，客户端终端同步打印反馈；
- 倒计时结束后，客户端打印结果序列
[0]，服务器打印“倒计时完成”； - CLI发送目标时，终端实时显示
Feedback字段的剩余秒数，最终显示Result。
常见问题
- 找不到动作服务：确认服务器已启动且
ros2 action list能看到/countdown；客户端需先调用wait_for_server()。 - 反馈不显示：客户端必须传入
feedback_callback参数，且服务器需调用publish_feedback()。 - CLI发送目标无反馈：需加
--feedback标志才会实时打印反馈字段。