4. ROS2动作通信基础

任务目标

本案例将实现一个ROS2动作通信的最小闭环：动作服务器执行一个可中断的长时间任务（模拟倒计时），动作客户端发送目标并实时接收进度反馈，最终获取执行结果。通过该案例掌握ROS2动作通信的三要素——目标、反馈、结果——以及对应的接口定义和编程模型。

环境与依赖

• Ubuntu 22.04 + ROS2 Humble（或更高版本）
• 已安装 ros-humble-ros2cli 和 ros-humble-example-interfaces（用于验证）
• 工作空间：~/ros2_ws/src

实现步骤

动作通信需要先定义自定义动作接口，然后分别实现服务器端和客户端节点。本案例使用Python编写，以最小代码量完成完整闭环。

1. 创建功能包并定义动作接口

在终端中执行以下命令创建功能包和动作目录：

cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python action_demo
mkdir -p action_demo/action

创建动作接口文件 action_demo/action/Counter.action：

# 目标：请求倒计时的总秒数
int32 total_seconds
---
# 结果：实际完成的倒计时秒数
int32 completed_seconds
---
# 反馈：当前剩余秒数
int32 remaining_seconds

编辑 CMakeLists.txt，在 find_package 部分添加 rosidl_default_generators，并在文件末尾追加：

find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
  "action/Counter.action"
)

编辑 package.xml，添加以下依赖声明：

<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>

2. 动作服务器节点

创建 action_demo/counter_server.py：

import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from action_demo.action import Counter

class CounterServer(Node):
    def __init__(self):
        super().__init__('counter_server')
        self._action_server = ActionServer(
            self, Counter, 'counter', self.execute_callback)

    def execute_callback(self, goal_handle):
        self.get_logger().info(f'收到目标: {goal_handle.request.total_seconds}秒')
        feedback_msg = Counter.Feedback()
        remaining = goal_handle.request.total_seconds

        while remaining > 0:
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                self.get_logger().info('目标被取消')
                return Counter.Result(completed_seconds=0)
            remaining -= 1
            feedback_msg.remaining_seconds = remaining
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'反馈: 剩余 {remaining} 秒')
            rclpy.spin_once(self, timeout_sec=1.0)

        goal_handle.succeed()
        result = Counter.Result(completed_seconds=goal_handle.request.total_seconds)
        self.get_logger().info('任务完成')
        return result

def main(args=None):
    rclpy.init(args=args)
    node = CounterServer()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

3. 动作客户端节点

创建 action_demo/counter_client.py：

import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from action_demo.action import Counter

class CounterClient(Node):
    def __init__(self):
        super().__init__('counter_client')
        self._action_client = ActionClient(self, Counter, 'counter')

    def send_goal(self, total_seconds):
        goal_msg = Counter.Goal()
        goal_msg.total_seconds = total_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.remaining_seconds} 秒')

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

def main(args=None):
    rclpy.init(args=args)
    node = CounterClient()
    node.send_goal(5)
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass

if __name__ == '__main__':
    main()

4. 配置入口并编译

编辑 setup.py，在 entry_points 中添加：

entry_points={
    'console_scripts': [
        'counter_server = action_demo.counter_server:main',
        'counter_client = action_demo.counter_client:main',
    ],
},

编译并刷新环境：

cd ~/ros2_ws
colcon build --packages-select action_demo
source install/setup.bash

运行与验证

打开三个终端，分别执行以下命令：

# 终端1：运行动作服务器
ros2 run action_demo counter_server

# 终端2：运行动作客户端（发送5秒倒计时目标）
ros2 run action_demo counter_client

# 终端3：查看动作通信状态
ros2 action list -t

验证通过条件：

1. 终端3显示动作类型 action_demo/action/Counter 及对应的服务器和客户端节点；
2. 终端2依次输出“目标被接受”和5条“收到反馈: 剩余 N 秒”（N从4递减到0）；
3. 终端1同步输出对应的反馈日志，最后输出“任务完成”；
4. 终端2最后输出“任务完成，实际执行: 5 秒”，随后节点自动退出。

常见问题

• 找不到动作接口：确认已执行 source install/setup.bash，且 CMakeLists.txt 中 rosidl_generate_interfaces 配置正确。
• 服务器不响应取消：本案例在循环中通过 is_cancel_requested 检查取消请求，若需测试取消功能，可在客户端调用 goal_handle.cancel_goal()。