4. 动作通信

任务目标

本案例实现一个ROS2动作通信（Action）的最小闭环：动作服务器执行倒计时任务（从指定秒数递减至0），动作客户端发送目标、接收周期性反馈，并支持中途取消。通过该案例掌握ROS2动作通信的三要素——目标（Goal）、反馈（Feedback）、结果（Result）——及其异步交互模型。

环境与依赖

操作系统：Ubuntu 22.04（或兼容ROS2环境）
ROS2发行版：Humble（或其他支持Python编写的发行版）
已安装ros-humble-ros2launch（通常随桌面版安装）

实现方案

动作接口使用ROS2标准示例接口example_interfaces/action/Fibonacci不合适（其语义为斐波那契数列），因此本案例自定义一个Countdown.action接口，包含：目标（倒计时秒数）、反馈（剩余秒数）、结果（是否完成）。服务器端使用rclpy.action.ActionServer，客户端使用rclpy.action.ActionClient。取消通过客户端调用cancel_goal()实现。

完整代码

以下代码包含三个文件：动作接口定义、服务器节点、客户端节点。将三个文件分别保存后，按后续步骤编译运行。

1. 动作接口定义（Countdown.action）

# Countdown.action
int32 seconds  # 目标：倒计时秒数
---
int32 remaining  # 反馈：剩余秒数
---
bool success  # 结果：是否完成

2. 动作服务器（action_server.py）

import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from rclpy.action.server import ServerGoalHandle
from rclpy.duration import Duration
from custom_action_interface.action import Countdown

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):
        # 接受所有目标（可在此添加合法性校验）
        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: ServerGoalHandle):
        self.get_logger().info(f'开始执行倒计时：{goal_handle.request.seconds}秒')
        feedback_msg = Countdown.Feedback()
        remaining = goal_handle.request.seconds

        while remaining > 0:
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                self.get_logger().info('目标已取消')
                return Countdown.Result(success=False)

            feedback_msg.remaining = remaining
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'反馈：剩余 {remaining} 秒')

            # 等待1秒，期间可响应取消
            for _ in range(10):
                if goal_handle.is_cancel_requested:
                    break
                await rclpy.spin_until_future_complete(
                    self,
                    rclpy.task.Future(),
                    timeout_sec=0.1
                )
            remaining -= 1

        goal_handle.succeed()
        self.get_logger().info('倒计时完成')
        return Countdown.Result(success=True)

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()

3. 动作客户端（action_client.py）

import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from custom_action_interface.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, seconds):
        goal_msg = Countdown.Goal()
        goal_msg.seconds = seconds
        self.get_logger().info(f'发送目标：倒计时 {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):
        remaining = feedback_msg.feedback.remaining
        self.get_logger().info(f'收到反馈：剩余 {remaining} 秒')

    def get_result_callback(self, future):
        result = future.result().result
        if result.success:
            self.get_logger().info('倒计时成功完成')
        else:
            self.get_logger().info('倒计时被取消')
        rclpy.shutdown()

def main(args=None):
    rclpy.init(args=args)
    client = CountdownClient()
    client.send_goal(5)  # 发送5秒倒计时目标

    # 3秒后发送取消请求（演示取消功能）
    def cancel_timer():
        client.get_logger().info('3秒已到，发送取消请求')
        client._action_client.cancel_goal(client._send_goal_future.result())

    client.create_timer(3.0, cancel_timer)
    rclpy.spin(client)

if __name__ == '__main__':
    main()

编译与运行

创建ROS2包并放置接口定义：

mkdir -p ~/ros2_action_ws/src
cd ~/ros2_action_ws/src
ros2 pkg create custom_action_interface --interface
# 将Countdown.action放入 custom_action_interface/action/ 目录

将两个Python文件放入custom_action_interface/同级包（或新建action_demo包），并确保setup.py中正确注册入口点：

entry_points={
    'console_scripts': [
        'server = action_demo.action_server:main',
        'client = action_demo.action_client:main',
    ],
},

编译并运行：

cd ~/ros2_action_ws
colcon build
source install/setup.bash
# 终端1：启动服务器
ros2 run action_demo server
# 终端2：启动客户端（自动发送5秒目标，3秒后取消）
ros2 run action_demo client

验证方法与通过标准

预期行为：服务器启动后等待目标；客户端发送5秒倒计时目标，服务器每秒发布一次剩余秒数反馈（5→4→3），客户端日志同步显示反馈；3秒时客户端发送取消请求，服务器响应取消并返回success=False，客户端打印“倒计时被取消”。
通过条件：终端日志中依次出现“发送目标”“目标被接受”“反馈：剩余5秒”“反馈：剩余4秒”“反馈：剩余3秒”“3秒已到，发送取消请求”“收到取消请求”“目标已取消”“倒计时被取消”，且无报错退出。
附加验证：若将客户端send_goal(5)改为send_goal(2)并注释掉取消定时器，则服务器完整执行2秒倒计时，客户端打印“倒计时成功完成”。

常见问题

找不到动作接口：确认Countdown.action已放入正确目录且执行过colcon build并source install/setup.bash。
取消不生效：服务器端execute_callback中必须周期性检查goal_handle.is_cancel_requested，本案例通过0.1秒轮询实现，确保取消请求在1秒内被响应。
客户端提前退出：客户端在get_result_callback中调用rclpy.shutdown()，若取消后未收到结果回调，可检查服务器是否正确调用了canceled()。

📚 课程目录