[ROS 系列学习教程] ROS动作通讯(Action)：通信模型、Hello World与拓展

ROS 系列学习教程(总目录)
ROS2 系列学习教程(总目录)

严格来说，Action不是基本通讯模型，它的底层由Topic组成。

Action通讯和Service类似，但单次Service通讯是阻塞的，只有服务端将服务请求处理完成后，才会返回结果给客户端。而有些场景中，我们不仅需要知道请求的处理结果，还需要知道请求处理的进度或其他中间结果。

比如，要控制机器人到达目标点A，Service通讯方式收到请求后会执行任务直到成功或失败，过程中客户端不会收到任何关于该任务的信息；而Action通讯方式，首先客户端告诉服务端机器人要走到A；服务端确认添加任务；然后服务端开始执行任务；最后服务端返回任务执行结果。

一、动作通讯模型

Action是一种用于处理长时间运行任务的通信机制，采用客户端-服务器模型，主要由5个Topic实现：

Topic 名称	消息类型	方向	作用
/action_name/goal	ActionNameActionGoal	Client → Server	发送目标
/action_name/cancel	actionlib_msgs/GoalID	Client → Server	取消请求
/action_name/status	actionlib_msgs/GoalStatusArray	Server → Client	状态更新
/action_name/feedback	ActionNameActionFeedback	Server → Client	进度反馈
/action_name/result	ActionNameActionResult	Server → Client	最终结果

二、动作通讯流程

主要分为添加、执行、完成三个阶段：

2.1 任务添加阶段

​ 由客户端向服务端发送目标的Topic，服务端接收后，向客户端发送确认的Topic

2.2 任务执行阶段

​ 目标任务确认后，服务端开始执行任务，并周期性向客户端发送任务执行信息

2.3 任务完成阶段

​ 任务成功、失败或取消，由服务端向客户端发送任务结果信息

三、Action Hello World

万物始于Hello World，同样，使用Hello World介绍Action的简单使用。

使用Action时，需要注意以下几点：

• Action名称
• 消息格式（.action，目标、最终结果、连续反馈）
• 客户端实现（发送目标，处理确认目标、定期任务信息、任务完成等各阶段的回调）
• 服务端实现（初始化服务器，发布定期任务信息，发布任务信息等）

为便于理解，我们使用Action实现一个查找n以内所有质数的功能。

3.1 创建并初始化功能包

（这一步不是必须，这里只是为了方便清晰的说明，也可以使用已有的包，在包里新增节点等方法）

首先创建 action_hello_world 包，命令如下：

catkin_create_pkg action_hello_world actionlib roscpp rospy

创建后，文件结构如下：

3.2 确定Action名称及消息格式

Action名称：/find_primes

消息文件名：FindPrimes.action

消息文件路径：在创建的 action_hello_world 包路径下创建一个 action 目录，将FindPrimes.action存放在该目录下

消息文件内容：

# 目标
int32 number # 查找 number 以内的质数
---
# 最终结果
int32 number 
int32[] primes # number 以内的质数
---
# 连续反馈
int32 number # 当前检查的数字
bool is_prime # 是否是质数

这里说明一下，Action的消息体由固定的 目标、最终结果、连续反馈 三部分组成，每一部分可以看做是一个msg消息体，依次对应前文 动作通讯流程 中的任务添加、完成、执行三个阶段。

3.3 配置编译文件

1. 添加message_generation功能包

message_generation功能包，在构建时根据action、 msg和srv生成消息和服务的接口文件（比如C++头文件和Python包），以便在 ROS 节点中使用。

在package.xml中添加以下内容：

<build_depend>message_generation</build_depend>

小知识：

这里其实也需要在 CMakeLists.txt 添加，类似如下：

find_package(catkin REQUIRED COMPONENTS
actionlib
roscpp
rospy
message_generation
)

但由于actionlib已经隐式调用过了message_generation，所以这里不需要显式声明。

2. 添加action文件

在 CMakeLists.txt 添加自定义action，该函数依赖message_generation功能包。

add_action_files(FILESFindPrimes.action
)

3. 配置依赖并生成接口文件

添加处理action、 msg或srv所需要的依赖，并生成接口文件，该函数依赖message_generation功能包。

generate_messages(DEPENDENCIESactionlib
)

4 . 添加message_runtime依赖

message_runtime 用于在运行时提供消息的序列化和反序列化支持。

这里注意，有时可能会看到没有显式添加 message_runtime 也能正常运行，这通常是因为其他依赖项（例如roscpp 或 std_msgs）可能已经隐含地包含了 message_runtime。在这种情况下，构建系统已经处理了消息生成的任务。

然而，为了确保你的软件包在所有情况下都能正常工作，最好显式添加 message_runtime 作为你的软件包的依赖项。这样可以确保你的消息定义在构建和运行时得到正确处理。

需要对 CMakeLists.txt 作以下修改：

catkin_package(CATKIN_DEPENDS roscpp rospy actionlib message_runtime
)

同时在package.xml中添加以下内容：

<exec_depend>message_runtime</exec_depend>

3.4 实现服务端与客户端（C++版）

在创建的 action_hello_world 包路径下有一个 src 目录，在这里存储C++源码，我们创建 action_hello_world_server.cpp 以实现服务端，编辑内容如下：

/*实现流程:1.包含头文件2.初始化 ROS 节点:命名(唯一)3.实例化 ROS 句柄4.实例化 Action服务器 对象5.实现服务任务（发布任务反馈、返回最终结果）6.启动 Action 服务器
*/// 1.包含头文件
#include <cmath>
#include <actionlib/server/simple_action_server.h>
#include <action_hello_world/FindPrimesAction.h>typedef actionlib::SimpleActionServer<action_hello_world::FindPrimesAction> Server;// 判断给定数字是否是质数
bool isPrime(int n)
{bool ret = true;if (n <= 1){ret = false;}else if (n % 2 == 0){ret = (n == 2);}else{for (int i = 3;; i += 2){if (i > n / i){break;}else if (n % i == 0){ret = false;}}}ROS_INFO("%d %s prime.", n, ret ? "is" : "is not");return ret;
}// 5.实现服务任务（发布任务反馈、返回最终结果）
void execute(const action_hello_world::FindPrimesGoalConstPtr &goal, Server *server)
{action_hello_world::FindPrimesFeedback feedback;std::vector<int> primes; // 存放找到的质数for (size_t i = 0; i <= goal->number; i++){feedback.number = i;feedback.is_prime = isPrime(i);server->publishFeedback(feedback); // 发布反馈// 存储找到的质数if (feedback.is_prime){primes.push_back(i);}ros::Duration(0.5).sleep();}// 返回最终结果action_hello_world::FindPrimesResult result;result.number = goal->number;result.primes = primes;server->setSucceeded(result);
}int main(int argc, char **argv)
{// 解决中文打印乱码setlocale(LC_ALL, "");// 2.初始化 ROS 节点:命名(唯一)ros::init(argc, argv, "action_hello_world_server");// 3.实例化 ROS 句柄ros::NodeHandle nh;// 4.实例化 Action服务器 对象Server server(nh, "/find_primes", boost::bind(&execute, _1, &server), false);// 6.启动服务器server.start();ros::spin();return 0;
}

创建 action_hello_world_client.cpp 以实现客户端，编辑内容如下：

/*实现流程:1.包含头文件2.初始化 ROS 节点:命名(唯一)3.实例化 Action客户端 对象4.等待服务端启动完成5.实例化目标对象6.发送目标任务到服务端7.等待任务完成
*/// 1.包含头文件
#include <ros/ros.h>
#include <sstream>
#include <actionlib/client/simple_action_client.h>
#include <action_hello_world/FindPrimesAction.h>typedef actionlib::SimpleActionClient<action_hello_world::FindPrimesAction> Client;// 处理最终结果
void done_cb(const actionlib::SimpleClientGoalState &state, const action_hello_world::FindPrimesResultConstPtr &result)
{if (state.state_ == state.SUCCEEDED){std::stringstream ss;for (const int n : result->primes){ss << std::to_string(n) << " ";}ROS_INFO("%d 以内的质数有: [ %s]", result->number, ss.str().c_str());}
}// 确认目标，服务激活
void active_cb()
{ROS_INFO("开始查找...");
}// 处理连续反馈的定期任务信息
void feedback_cb(const action_hello_world::FindPrimesFeedbackConstPtr &feedback)
{ROS_INFO("当前数字: %d, %s质数", feedback->number, feedback->is_prime?"是":"不是");
}int main(int argc, char **argv)
{setlocale(LC_ALL, ""); // 解决中文打印乱码// 2.初始化 ROS 节点:命名(唯一)ros::init(argc, argv, "action_hello_world_client");// 3.实例化 Action客户端 对象Client client("/find_primes", true);// 4.等待服务端启动完成client.waitForServer();ROS_INFO("添加任务前的状态: %s", client.getState().toString().c_str());// 5.实例化目标对象action_hello_world::FindPrimesGoal goal;goal.number = 12;ROS_INFO("查找 %d 以内的质数", goal.number);// 6.发送目标任务到服务端，并注册任务完成、确认目标、定期任务信息的回调函数client.sendGoal(goal, &done_cb, &active_cb, &feedback_cb);ROS_INFO("添加任务后的状态: %s", client.getState().toString().c_str());ros::Duration(1.0).sleep();ROS_INFO("执行任务时的状态: %s", client.getState().toString().c_str());// 7.等待任务完成client.waitForResult(ros::Duration(1000.0)); // 等待结果，1000秒超时ROS_INFO("任务执行完的状态: %s", client.getState().toString().c_str());return 0;
}

编译运行

进入工作空间执行 catkin_make 命令编译工程，编译成功后，使用如下命令依次启动服务器和客户端。

1. 启动ros master
roscore
2. 启动服务器
rosrun action_hello_world action_hello_world_server
3. 启动客户端
rosrun action_hello_world action_hello_world_client

结果如下：

目前为止，Action Hello World 已经成功了。

3.5 实现服务端与客户端（Python版）

在创建的 action_hello_world 包路径下 src 目录的同级，创建一个 scripts 目录，在这里存储脚本（如python脚本），我们创建 action_hello_world_server.py 以实现服务端，编辑内容如下：

#!/usr/bin/env python
# -*- coding: utf-8 -*-# 1. 导入依赖包
import rospy
import math
import actionlib
from action_hello_world.msg import FindPrimesAction, FindPrimesResult, FindPrimesFeedback# 判断给定数字是否是质数
def is_prime(n):if n <= 1:rospy.loginfo(f"{n} 不是质数")return Falseif n == 2:rospy.loginfo(f"{n} 是质数")return Trueif n % 2 == 0:rospy.loginfo(f"{n} 不是质数")return False# 检查从3到sqrt(n)的奇数因子ret = Truefor i in range(3, int(math.sqrt(n)) + 1, 2):if n % i == 0:ret = Falsebreakstatus_str = "是" if ret else "不是"rospy.loginfo(f"{n} {status_str}质数")return ret# 实现服务任务（发布任务反馈、返回最终结果）
def execute_cb(goal):primes = []  # 存放找到的质数target = goal.number# 创建反馈对象feedback = FindPrimesFeedback()# 从2开始检查到目标数字for num in range(2, target + 1):# 更新反馈信息feedback.number = numfeedback.is_prime = is_prime(num)# 如果是质数则添加到结果列表if feedback.is_prime:primes.append(num)# 发布反馈server.publish_feedback(feedback)# 模拟处理时间rospy.sleep(0.5)# 返回最终结果result = FindPrimesResult()result.number = targetresult.primes = primesserver.set_succeeded(result)if __name__ == "__main__":# 2. 初始化ROS节点rospy.init_node("action_hello_world_server")# 3. 实例化Action服务器对象server = actionlib.SimpleActionServer("/find_primes", FindPrimesAction, execute_cb, auto_start=False)# 4. 启动服务器server.start()rospy.loginfo("质数查找服务器已启动...")# 保持节点运行rospy.spin()

创建 action_hello_world_client.py 以实现客户端，编辑内容如下：

#!/usr/bin/env python
# -*- coding: utf-8 -*-# 1. 导入依赖包
import rospy
import actionlib
from action_hello_world.msg import FindPrimesAction, FindPrimesGoal, FindPrimesResult, FindPrimesFeedbackdef done_cb(state, result):"""处理最终结果回调"""if state == actionlib.GoalStatus.SUCCEEDED:primes_str = " ".join(str(n) for n in result.primes)rospy.loginfo(f"{result.number} 以内的质数有: [ {primes_str} ]")def active_cb():"""目标激活回调"""rospy.loginfo("开始查找...")def feedback_cb(feedback):"""处理反馈回调"""status_str = "是" if feedback.is_prime else "不是"rospy.loginfo(f"当前数字: {feedback.number}, {status_str}质数")if __name__ == "__main__":# 2. 初始化ROS节点rospy.init_node("action_hello_world_client")# 3. 实例化Action客户端对象client = actionlib.SimpleActionClient("/find_primes", FindPrimesAction)# 4. 等待服务端启动rospy.loginfo("等待服务端启动...")client.wait_for_server()rospy.loginfo(f"添加任务前的状态: {client.get_state()}")# 5. 实例化目标对象goal = FindPrimesGoal()goal.number = 12rospy.loginfo(f"查找 {goal.number} 以内的质数")# 6. 发送目标任务client.send_goal(goal, done_cb=done_cb, active_cb=active_cb, feedback_cb=feedback_cb)rospy.loginfo(f"添加任务后的状态: {client.get_state()}")rospy.sleep(1.0)rospy.loginfo(f"执行任务时的状态: {client.get_state()}")# 7. 等待任务完成client.wait_for_result(rospy.Duration(1000.0))  # 1000秒超时rospy.loginfo(f"任务执行完的状态: {client.get_state()}")

修改 CMakeLists.txt ，只需添加如下内容：

catkin_install_python(PROGRAMSscripts/action_hello_world_server.pyscripts/action_hello_world_client.pyDESTINATION ${CATKIN_PACKAGE_BIN_DESTINATION}
)

编译运行

进入工作空间执行 catkin_make 命令编译工程，编译成功后，使用如下命令依次启动服务端和客户端。

1. 启动ros master(如果已启动，无需再启动)
roscore
2. 启动服务端
rosrun action_hello_world action_hello_world_server.py
3. 启动客户端
rosrun action_hello_world action_hello_world_client.py

结果如下：