ROS2 话题、服务、动作通讯入门：Topic、Service、Action 一篇讲清楚

在 ROS2 中，节点之间主要通过三种方式通信：

• Topic：话题通信，适合连续数据流。
• Service：服务通信，适合一次请求、一次响应。
• Action：动作通信，适合耗时任务，并且可以持续反馈进度。

简单理解：

• Topic：我一直发，你需要就订阅。
• Service：你问我答。
• Action：你给我一个目标，我边做边反馈，做完给结果。

本文使用 Python rclpy 做简单示例。

一、Topic 话题通信

Topic 是 ROS2 中最常见的通信方式，采用发布/订阅模型。

发布者只负责向某个话题发送消息，订阅者只负责监听这个话题。发布者和订阅者不需要知道对方是谁。

典型场景：

• 传感器数据发布
• 机器人状态发布
• 速度指令下发
• 电量、位姿、关节状态上报

Topic 适合连续、周期性、不要求回应的数据。

1. 发布节点

下面的代码每 0.5 秒向 /chatter 话题发布一条字符串消息。

import rclpy  									# 导入 ROS2 Python 客户端库
from rclpy.node import Node  					# 导入 ROS2 节点基类
from std_msgs.msg import String  				# 导入字符串消息类型


class SimplePublisher(Node):
    def __init__(self):
        super().__init__("simple_publisher")    # 初始化节点，节点名为 simple_publisher
        self.publisher = self.create_publisher(String, "/chatter", 10)  	# 创建发布者，参数：消息类型、话题名、队列长度
        self.count = 0  # 计数器，用来生成不同消息内容
        self.timer = self.create_timer(0.5, self.publish_message)  # 创建定时器，每 0.5 秒调用一次发布函数

    def publish_message(self):
        msg = String()  								# 创建 String 消息对象
        msg.data = f"hello ros2: {self.count}"  		# 给消息内容赋值
        self.publisher.publish(msg)  					# 发布消息到 /chatter 话题
        self.get_logger().info(f"publish: {msg.data}")  # 打印发布日志
        self.count += 1  								# 计数器自增


def main():
    rclpy.init()  										# 初始化 ROS2
    node = SimplePublisher()  							# 创建发布节点
    rclpy.spin(node)  									# 让节点持续运行
    node.destroy_node()  								# 销毁节点
    rclpy.shutdown()  									# 关闭 ROS2


if __name__ == "__main__":
    main()

2. 订阅节点

下面的代码订阅 /chatter 话题，收到消息后打印内容。

import rclpy  									# 导入 ROS2 Python 客户端库
from rclpy.node import Node 					# 导入 ROS2 节点基类
from std_msgs.msg import String 			    # 导入字符串消息类型


class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__("simple_subscriber")   # 初始化节点，节点名为 simple_subscriber
        self.subscription = self.create_subscription(
            String,  							# 消息类型，必须和发布端一致
            "/chatter",  						# 订阅的话题名，必须和发布端一致
            self.listener_callback, 			# 收到消息后执行的回调函数
            10,  								# 队列长度
        )

    def listener_callback(self, msg):
        self.get_logger().info(f"receive: {msg.data}")  	# 打印接收到的消息内容


def main():
    rclpy.init()  											# 初始化 ROS2
    node = SimpleSubscriber() 								# 创建订阅节点
    rclpy.spin(node) 										# 让节点持续运行，等待接收消息
    node.destroy_node()  									# 销毁节点
    rclpy.shutdown()  										# 关闭 ROS2


if __name__ == "__main__":
    main()

3. Topic 特点

• 适合连续数据流。
• 发布者和订阅者解耦。
• 一个话题可以有多个发布者和多个订阅者。
• 发布者不等待订阅者回复。

二、Service 服务通信

Service 是请求/响应模型。

客户端发送一个请求，服务端处理后返回一个响应。它适合短时间完成的任务。

典型场景：

• 查询机器人状态
• 触发一次拍照
• 保存地图
• 切换模式
• 读取或修改参数

这里使用 ROS2 自带的 std_srvs/srv/Trigger 类型。它的请求为空，响应包含 success 和 message。

1. 服务端

import rclpy  										# 导入 ROS2 Python 客户端库
from rclpy.node import Node  						# 导入 ROS2 节点基类
from std_srvs.srv import Trigger  					# 导入 Trigger 服务类型


class HealthService(Node):
    def __init__(self):
        super().__init__("health_service")  		# 初始化服务端节点
        self.service = self.create_service(
            Trigger,  								# 服务类型
            "/health_check", 						# 服务名
            self.handle_request,  					# 收到请求后执行的回调函数
        )

    def handle_request(self, request, response):
        response.success = True  					# 设置响应结果为成功
        response.message = "robot is healthy"  		# 设置响应信息
        self.get_logger().info("receive health check request")  # 打印请求日志
        return response  										# 返回响应给客户端


def main():
    rclpy.init()  # 初始化 ROS2
    node = HealthService()  # 创建服务端节点
    rclpy.spin(node)  # 让服务端持续运行，等待客户端请求
    node.destroy_node()  # 销毁节点
    rclpy.shutdown()  # 关闭 ROS2


if __name__ == "__main__":
    main()

2. 客户端

import rclpy  									# 导入 ROS2 Python 客户端库
from rclpy.node import Node  					# 导入 ROS2 节点基类
from std_srvs.srv import Trigger 				# 导入 Trigger 服务类型


class HealthClient(Node):
    def __init__(self):
        super().__init__("health_client")  		# 初始化客户端节点
        self.client = self.create_client(Trigger, "/health_check")  # 创建服务客户端，参数：服务类型、服务名

    def send_request(self):
        if not self.client.wait_for_service(timeout_sec=5.0):  # 等待服务端上线，最多等待 5 秒
            self.get_logger().error("service not available")   # 服务不可用时打印错误
            return

        request = Trigger.Request()  					# 创建请求对象，Trigger 请求内容为空
        future = self.client.call_async(request) 		# 异步发送请求
        rclpy.spin_until_future_complete(self, future)  # 等待服务端响应

        response = future.result()  					# 获取响应结果
        self.get_logger().info(
            f"success={response.success}, message={response.message}"  # 打印响应内容
        )


def main():
    rclpy.init()  										# 初始化 ROS2
    node = HealthClient()  								# 创建客户端节点
    node.send_request() 								# 发送一次服务请求
    node.destroy_node()  								# 销毁节点
    rclpy.shutdown()  									# 关闭 ROS2


if __name__ == "__main__":
    main()

3. Service 特点

• 一次请求对应一次响应。
• 适合低频、短耗时任务。
• 不适合连续数据流。
• 不适合长时间任务。

三、Action 动作通信

Action 用于长时间执行的任务。它比 Service 多了反馈和取消机制。

Action 通常包含三部分：

• Goal：客户端发送目标。
• Feedback：服务端执行过程中返回反馈。
• Result：任务完成后返回最终结果。

典型场景：

• 导航到目标点
• 机械臂移动到目标姿态
• 长时间路径规划
• 需要进度反馈或取消的任务

下面使用 ROS2 自带的 example_interfaces/action/Fibonacci 做演示。

1. Action 服务端

import time  										# 用于模拟耗时任务

import rclpy 										# 导入 ROS2 Python 客户端库
from rclpy.node import Node  						# 导入 ROS2 节点基类
from rclpy.action import ActionServer  				# 导入 Action 服务端类
from example_interfaces.action import Fibonacci     # 导入 Fibonacci Action 类型


class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__("fibonacci_action_server")  # 初始化 Action 服务端节点
        self.action_server = ActionServer(
            self,  									 # 当前节点
            Fibonacci,  							 # Action 类型
            "/fibonacci",  							 # Action 名称
            self.execute_callback,  				 # 收到目标后执行的回调函数
        )

    def execute_callback(self, goal_handle):
        order = goal_handle.request.order  			# 获取客户端发送的目标值
        feedback = Fibonacci.Feedback() 		 	# 创建反馈对象
        feedback.sequence = [0, 1]  				# 初始化 Fibonacci 序列

        for i in range(2, order):  					# 从第 3 个数开始计算
            if goal_handle.is_cancel_requested:  	# 判断客户端是否请求取消任务
                goal_handle.canceled()  			# 标记任务已取消
                result = Fibonacci.Result()  		 # 创建结果对象
                result.sequence = feedback.sequence  # 返回当前已经计算出的序列
                return result

            next_value = feedback.sequence[-1] + feedback.sequence[-2]  # 计算下一个 Fibonacci 数
            feedback.sequence.append(next_value)  					# 更新反馈序列
            goal_handle.publish_feedback(feedback)  				# 发送执行过程中的反馈
            time.sleep(0.5)  										# 模拟耗时计算

        goal_handle.succeed()  										# 标记任务执行成功
        result = Fibonacci.Result()  								# 创建最终结果对象
        result.sequence = feedback.sequence 		 				# 设置最终结果
        return result  												# 返回最终结果给客户端


def main():
    rclpy.init()  											# 初始化 ROS2
    node = FibonacciActionServer()  						# 创建 Action 服务端节点
    rclpy.spin(node)  										# 持续运行，等待客户端发送目标
    node.destroy_node() 					 				# 销毁节点
    rclpy.shutdown()  										# 关闭 ROS2


if __name__ == "__main__":
    main()

2. Action 客户端

import rclpy  									 # 导入 ROS2 Python 客户端库
from rclpy.node import Node  					 # 导入 ROS2 节点基类
from rclpy.action import ActionClient  			 # 导入 Action 客户端类
from example_interfaces.action import Fibonacci  # 导入 Fibonacci Action 类型


class FibonacciActionClient(Node):
    def __init__(self):
        super().__init__("fibonacci_action_client") 		 # 初始化 Action 客户端节点
        self.action_client = ActionClient(self, Fibonacci, "/fibonacci")  # 创建 Action 客户端

    def send_goal(self):
        if not self.action_client.wait_for_server(timeout_sec=5.0):  # 等待 Action 服务端上线
            self.get_logger().error("action server not available")  # 服务端不可用时打印错误
            return

        goal = Fibonacci.Goal()  					# 创建目标对象
        goal.order = 8  							# 设置目标：计算 8 个 Fibonacci 数

        future = self.action_client.send_goal_async(
            goal,  # 发送的目标
            feedback_callback=self.feedback_callback,  # 注册反馈回调函数
        )
        rclpy.spin_until_future_complete(self, future)  # 等待目标发送完成

        goal_handle = future.result()  					# 获取目标句柄
        if not goal_handle.accepted: 			 		# 判断目标是否被服务端接受
            self.get_logger().error("goal rejected")    # 目标被拒绝时打印错误
            return

        result_future = goal_handle.get_result_async()  # 异步等待最终结果
        rclpy.spin_until_future_complete(self, result_future)  # 等待结果返回

        result = result_future.result().result  				# 获取最终结果
        self.get_logger().info(f"result: {result.sequence}")    # 打印最终结果

    def feedback_callback(self, feedback_msg):
        feedback = feedback_msg.feedback  						# 获取反馈内容
        self.get_logger().info(f"feedback: {feedback.sequence}")  # 打印执行过程中的反馈


def main():
    rclpy.init()  								# 初始化 ROS2
    node = FibonacciActionClient()  			# 创建 Action 客户端节点
    node.send_goal()  							# 发送目标
    node.destroy_node()  						# 销毁节点
    rclpy.shutdown()  							# 关闭 ROS2


if __name__ == "__main__":
    main()

3. Action 特点

• 适合耗时任务。
• 可以持续接收反馈。
• 可以返回最终结果。
• 可以取消任务。
• 比 Service 更适合机器人动作执行。

四、三种通信方式对比

通信方式	模型	适合场景	是否持续通信	是否有反馈	是否适合长任务
Topic	发布/订阅	传感器、状态、遥测	是	否	否
Service	请求/响应	查询状态、触发操作	否	否	否
Action	目标/反馈/结果	导航、机械臂、长任务	执行期间持续	是	是

五、运行方式

如果这些文件已经写入 ROS2 Python 包，并在 setup.py 中配置入口，可以按下面方式运行。

编译工作区：

cd ~/ros2_ws
source /opt/ros/jazzy/setup.bash
colcon build --symlink-install
source install/setup.bash

运行 Topic：

ros2 run your_package simple_subscriber
ros2 run your_package simple_publisher

运行 Service：

ros2 run your_package health_service
ros2 run your_package health_client

运行 Action：

ros2 run your_package fibonacci_action_server
ros2 run your_package fibonacci_action_client

六、总结

ROS2 的通信机制可以看作机器人系统内部的“语言”。

• 数据需要持续发送：用 Topic。
• 操作很快完成，并且需要立即返回结果：用 Service。
• 操作比较耗时，并且需要进度反馈或取消功能：用 Action。

掌握 Topic、Service、Action，就已经具备搭建基本 ROS2 多节点系统的能力了。