4. ROS2动作通信实战

任务目标

本案例实现一个ROS2动作通信（Action）的最小闭环：动作服务器执行一个模拟长时间任务（例如持续5秒的倒计时），动作客户端异步发送目标、接收反馈并获取最终结果。通过该案例掌握ROS2 Action的接口定义、服务器/客户端编写及异步执行模型。

环境与依赖

Ubuntu 22.04 + ROS2 Humble（或更高版本）
已安装 ros-humble-ros2cli（含action命令）
已source ROS2环境：source /opt/ros/humble/setup.bash

实现步骤

首先创建自定义Action接口，然后分别实现服务器节点和客户端节点，最后编译并运行验证。

1. 创建Action接口

在功能包中定义Action接口。创建文件 action/Countdown.action：

# 目标：倒计时总秒数
int32 total_time
---
# 结果：实际完成的倒计时秒数
int32 completed_time
---
# 反馈：剩余秒数
int32 remaining_time

在 CMakeLists.txt 中添加编译依赖（rosidl_generate_interfaces）并在 package.xml 中添加 rosidl_default_generators 与 action_msgs 依赖。此处假设功能包名为 action_demo。

2. 动作服务器代码

创建 action_server.py，实现服务器端逻辑：

#!/usr/bin/env python3
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from action_demo.action import Countdown
import time

class CountdownServer(Node):
    def __init__(self):
        super().__init__('countdown_server')
        self._action_server = ActionServer(
            self,
            Countdown,
            'countdown',
            execute_callback=self.execute_callback,
            goal_callback=self.goal_callback,
            cancel_callback=self.cancel_callback
        )
        self.get_logger().info('动作服务器已启动')

    def goal_callback(self, goal_request):
        # 接受所有目标
        self.get_logger().info(f'收到目标: {goal_request.total_time}秒')
        return rclpy.action.GoalResponse.ACCEPT

    def cancel_callback(self, goal_handle):
        self.get_logger().info('收到取消请求')
        return rclpy.action.CancelResponse.ACCEPT

    async def execute_callback(self, goal_handle):
        self.get_logger().info('开始执行倒计时...')
        feedback_msg = Countdown.Feedback()
        total = goal_handle.request.total_time
        remaining = total

        for i in range(total, 0, -1):
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                self.get_logger().info('任务已取消')
                return Countdown.Result(completed_time=total - remaining)

            remaining = i
            feedback_msg.remaining_time = remaining
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'反馈: 剩余 {remaining} 秒')
            time.sleep(1.0)

        goal_handle.succeed()
        result = Countdown.Result()
        result.completed_time = total
        self.get_logger().info('任务完成')
        return result

def main(args=None):
    rclpy.init(args=args)
    node = CountdownServer()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

3. 动作客户端代码

创建 action_client.py，实现异步发送目标并处理反馈：

#!/usr/bin/env python3
import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from action_demo.action import Countdown

class CountdownClient(Node):
    def __init__(self):
        super().__init__('countdown_client')
        self._action_client = ActionClient(self, Countdown, 'countdown')
        self.get_logger().info('动作客户端已启动')

    def send_goal(self, total_time):
        goal_msg = Countdown.Goal()
        goal_msg.total_time = total_time
        self.get_logger().info(f'发送目标: {total_time}秒')

        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):
        feedback = feedback_msg.feedback
        self.get_logger().info(f'收到反馈: 剩余 {feedback.remaining_time} 秒')

    def get_result_callback(self, future):
        result = future.result().result
        self.get_logger().info(f'任务完成，实际执行: {result.completed_time}秒')
        rclpy.shutdown()

def main(args=None):
    rclpy.init(args=args)
    node = CountdownClient()
    node.send_goal(5)  # 请求5秒倒计时
    rclpy.spin(node)

if __name__ == '__main__':
    main()

4. 编译与运行

在功能包根目录执行编译，然后分别运行服务器和客户端（两个终端）：

# 终端1：编译并source
colcon build --packages-select action_demo
source install/setup.bash
ros2 run action_demo action_server

# 终端2：source后运行客户端
source install/setup.bash
ros2 run action_demo action_client

验证方法

运行后观察以下现象以确认动作通信正常：

服务器终端显示“收到目标: 5秒”和“开始执行倒计时...”，随后每秒输出一次“反馈: 剩余 N 秒”（N从5递减到1）。
客户端终端同步显示“目标被接受”和每秒的“收到反馈: 剩余 N 秒”。
5秒后服务器输出“任务完成”，客户端输出“任务完成，实际执行: 5秒”，随后客户端自动退出。

预期输出示例（客户端终端）：

[INFO] 发送目标: 5秒
[INFO] 目标被接受
[INFO] 收到反馈: 剩余 5 秒
[INFO] 收到反馈: 剩余 4 秒
[INFO] 收到反馈: 剩余 3 秒
[INFO] 收到反馈: 剩余 2 秒
[INFO] 收到反馈: 剩余 1 秒
[INFO] 任务完成，实际执行: 5秒

该输出由代码逻辑直接推出，用于确认执行流程正确。若需验证取消功能，可在服务器运行期间向客户端发送Ctrl+C，观察服务器输出“任务已取消”。

常见问题

找不到action_demo包：确认已执行 colcon build 并source了 install/setup.bash。
Action类型未注册：检查 CMakeLists.txt 中是否包含 rosidl_generate_interfaces 调用，且包名与import路径一致。
客户端卡在wait_for_server：确认服务器已启动且Action名称（countdown）在两端一致。

📚 课程目录