ROS2核心概念之服务

话题通信可以实现多个ROS节点之间数据的单向传输，使用这种异步通信机制，发布者无法准确知道订阅者是否收到消息，本节我们将一起学习ROS另外一种常用的通信方法——服务，可以实现类似你问我答的同步通信效果。
一、 通信模型

在话题章节中，我们通过一个节点驱动相机，发布图像话题，另外一个节点订阅图像话题，并实现对其中橙色物体的识别，此时我们可以按照图像识别的频率，周期得到物体在图片中的位置。

这个位置信息可以继续发给机器人的上层应用使用，比如可以跟随目标运动，或者运动到目标位置附近。然而，有些场景我们并不需要这么高的频率一直订阅物体的位置，而是更希望在需要这个数据的时候，发一个查询的请求，然后尽快得到此时目标的最新位置。
这样的通信模型和话题单向传输有所不同，变成了发送一个请求，反馈一个应答的形式，好像是你问我答一样，这种通信机制在ROS中成为服务，Service。
1.1 客户端/服务器模型

从服务的实现机制上来看，这种你问我答的形式叫做客户端/服务器模型，简称为CS模型，客户端在需要某些数据的时候，针对某个具体的服务，发送请求信息，服务器端收到请求之后，就会进行处理并反馈应答信息。

这种通信机制在生活中也很常见，比如我们经常浏览的各种网页，此时你的电脑浏览器就是客户端，通过域名或者各种操作，向网站服务器发送请求，服务器收到之后返回需要展现的页面数据。
1.2 同步通信

这个过程一般要求越快越好，假设服务器半天没有反应，你的浏览器一直转圈圈，那有可能是服务器宕机了，或者是网络不好，所以相比话题通信，在服务通信中，客户端可以通过接收到的应答信息，判断服务器端的状态，我们也称之为同步通信。
1.3 一对多通信

比如博客园这个网站，服务器是唯一存在的，并没有多个完全一样的博客园网站，但是可以访问博客园网站的客户端是不唯一的，大家每一个人都可以看到同样的界面。所以服务通信模型中，服务器端唯一，但客户端可以不唯一。
1.4 服务接口

和话题通信类似，服务通信的核心还是要传递数据，数据变成了两个部分：

• 一个请求的数据，比如请求橘子位置的命令；
  
• 还有一个反馈的数据，比如反馈橘子坐标位置的数据；
  

这些数据和话题消息一样，在ROS中也是要标准定义的，话题使用.msg文件定义，服务使用的是.srv文件定义，后续我们会介绍定义的方法。
二、服务案例

接下来，我们就来看看， 服务到底该如何实现。创建my_learning_service的Python版本的功能包；

1. pi@NanoPC-T6:~/dev_ws$ cd src
2. pi@NanoPC-T6:~/dev_ws/src$ ros2 pkg create --build-type ament_python my_learning_service

复制代码

2.1 加法求解器

前面我们对ROS服务通信应该有了基本了解，接下来我们就要开始编写代码啦。还是从一个相对简单的例程开始，也是ROS官方的一个例程，通过服务实现一个加法求解器的功能。

当我们需要计算两个加数的求和结果时，我们应该怎么做呢？我们需要两个实现节点：

• 客户端节点：将两个加数封装成请求数据，针对服务add_two_ints发送出去；
  
• 服务器端节点：收到客户端发送的请求数据后，开始进行加法计算，并将求和结果封装成应答数据反馈给客户端。
  

2.1.1 客户端

使用VS Code加载功能包my_learning_service，在my_learning_service文件夹下创建service_adder_client.py；

1. """
2. ROS2服务示例-发送两个加数，请求加法器计算
4. @author: zy
5. @since : 2025/12/12
6. """
8. import sys
10. import rclpy                                     # ROS2 Python接口库
11. from rclpy.node   import Node                    # ROS2 节点类
12. from my_learning_interface.srv import AddTwoInts # 自定义的服务接口
14. class adderClient(Node):
15.     def __init__(self, name):
16.         super().__init__(name)                                       # ROS2节点父类初始化
17.         self.client = self.create_client(AddTwoInts, 'add_two_ints') # 创建服务客户端对象（服务接口类型，服务名）
18.         while not self.client.wait_for_service(timeout_sec=1.0):     # 循环等待服务器端成功启动
19.             self.get_logger().info('service not available, waiting again...')
20.         self.request = AddTwoInts.Request()                          # 创建服务请求的数据对象
22.     def send_request(self):                                          # 创建一个发送服务请求的函数
23.         self.request.a = int(sys.argv[1])
24.         self.request.b = int(sys.argv[2])
25.         self.future = self.client.call_async(self.request)           # 异步方式发送服务请求
27. def main(args=None):
28.     rclpy.init(args=args)                        # ROS2 Python接口初始化
29.     node = adderClient("service_adder_client")   # 创建ROS2节点对象并进行初始化
30.     node.send_request()                          # 发送服务请求
32.     while rclpy.ok():                            # ROS2系统正常运行
33.         rclpy.spin_once(node)                    # 循环执行一次节点，看看服务是不是返回了
35.         if node.future.done():                   # 数据是否处理完成，即服务数据返回了
36.             try:
37.                 response = node.future.result()  # 接收服务器端的反馈数据
38.             except Exception as e:
39.                 node.get_logger().info(
40.                     'Service call failed %r' % (e,))
41.             else:
42.                 node.get_logger().info(          # 将收到的反馈信息打印输出
43.                     'Result of add_two_ints: for %d + %d = %d' %
44.                     (node.request.a, node.request.b, response.sum))
45.             break
47.     node.destroy_node()                          # 销毁节点对象
48.     rclpy.shutdown()                             # 关闭ROS2 Python接口

复制代码

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

1.     entry_points={
2.         'console_scripts': [
3.          'service_adder_client  = my_learning_service.service_adder_client:main',
4.         ],
5.     },

复制代码

对以上程序进行分析，如果我们想要实现一个客户端，流程如下：

• 编程接口初始化；
  
• 创建节点并初始化；
  
• 创建客户端对象；
  
• 创建并发送请求数据；
  
• 等待服务器端应答数据；
  
• 销毁节点并关闭接口；
  
• 服务端代码解析。
  

2.1.2 服务端

在my_learning_service文件夹下创建service_adder_server.py；

1. """
2. ROS2服务示例-提供加法器的服务器处理功能
4. @author: zy
5. @since : 2025/12/12
6. """
8. import rclpy                                        # ROS2 Python接口库
9. from rclpy.node   import Node                       # ROS2 节点类
10. from my_learning_interface.srv import AddTwoInts    # 自定义的服务接口
12. class adderServer(Node):
13.     def __init__(self, name):
14.         super().__init__(name)                                                           # ROS2节点父类初始化
15.         self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.adder_callback)  # 创建服务器对象（接口类型、服务名、服务器回调函数）
17.     def adder_callback(self, request, response):   # 创建回调函数，执行收到请求后对数据的处理
18.         response.sum = request.a + request.b       # 完成加法求和计算，将结果放到反馈的数据中
19.         self.get_logger().info('Incoming request\na: %d b: %d' % (request.a, request.b))   # 输出日志信息，提示已经完成加法求和计算
20.         return response                          # 反馈应答信息
22. def main(args=None):                             # ROS2节点主入口main函数
23.     rclpy.init(args=args)                        # ROS2 Python接口初始化
24.     node = adderServer("service_adder_server")   # 创建ROS2节点对象并进行初始化
25.     rclpy.spin(node)                             # 循环等待ROS2退出
26.     node.destroy_node()                          # 销毁节点对象
27.     rclpy.shutdown()                             # 关闭ROS2 Python接口

复制代码

服务器端的实现，有点类似话题通信中的订阅者，并不知道请求数据什么时间出现，也用到了回调函数机制。
完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

1.     entry_points={
2.         'console_scripts': [
3.          'service_adder_client  = my_learning_service.service_adder_client:main',
4.          'service_adder_server  = my_learning_service.service_adder_server:main',
5.         ],
6.     },

复制代码

对以上程序进行分析，如果我们想要实现一个服务端，流程如下：

• 编程接口初始化；
  
• 创建节点并初始化；
  
• 创建服务器端对象；
  
• 通过回调函数处进行服务；
  
• 向客户端反馈应答结果；
  
• 销毁节点并关闭接口。
  

2.1.3 编译运行

编译程序：

1. pi@NanoPC-T6:~/dev_ws$ colcon build --paths src/my_learning_service

复制代码

启动第一个终端，第一个节点是服务端，等待请求数据并提供求和功能；

1. pi@NanoPC-T6:~/dev_ws$ ros2 run my_learning_service service_adder_server

复制代码

启动第二个终端，第二个节点是客户端，发送传入的两个加数并等待求和结果；

1. pi@NanoPC-T6:~/dev_ws$ ros2 run my_learning_service service_adder_client 2 3

复制代码

2.2 机器视觉识别

好啦，加法求解器已经实现了，回想下刚才我们提到的视觉识别流程，当我们需要知道目标物体位置的时候，通过服务通信的机制，岂不是更加合理。

接着我们采用服务通信的机制对物体位置识别进行改造，此时会有三个节点出现：

• 相机驱动节点，发布图像数据；
  
• 视觉识别节点，订阅图像数据，并且集成了一个服务器端对象，随时准备提供目标位置；
  
• 客户端节点，我们可以认为是一个机器人目标跟踪的节点，当需要根据目标运动时，就发送一次请求，然后拿到一个当前的目标位置。
  

2.2.1 客户端

在my_learning_service文件夹下创建service_object_client.py；

1. """
2. ROS2服务示例-请求目标识别，等待目标位置应答
4. @author: zy
5. @since : 2025/12/12
6. """
8. import rclpy                                            # ROS2 Python接口库
9. from rclpy.node import Node                             # ROS2 节点类
10. from my_learning_interface.srv import GetObjectPosition # 自定义的服务接口
12. class objectClient(Node):
13.     def __init__(self, name):
14.         super().__init__(name)                          # ROS2节点父类初始化
15.         self.client = self.create_client(GetObjectPosition, 'get_target_position')
16.         while not self.client.wait_for_service(timeout_sec=1.0):
17.             self.get_logger().info('service not available, waiting again...')
18.         self.request = GetObjectPosition.Request()
20.     def send_request(self):
21.         self.request.get = True
22.         self.future = self.client.call_async(self.request)
24. def main(args=None):
25.     rclpy.init(args=args)                             # ROS2 Python接口初始化
26.     node = objectClient("service_object_client")      # 创建ROS2节点对象并进行初始化
27.     node.send_request()
29.     while rclpy.ok():
30.         rclpy.spin_once(node)
32.         if node.future.done():
33.             try:
34.                 response = node.future.result()
35.             except Exception as e:
36.                 node.get_logger().info(
37.                     'Service call failed %r' % (e,))
38.             else:
39.                 node.get_logger().info(
40.                     'Result of object position:\n x: %d y: %d' %
41.                     (response.x, response.y))
42.             break
43.     node.destroy_node()                              # 销毁节点对象
44.     rclpy.shutdown()                                 # 关闭ROS2 Python接口

复制代码

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

1.     entry_points={
2.         'console_scripts': [
3.          'service_adder_client  = my_learning_service.service_adder_client:main',
4.          'service_adder_server  = my_learning_service.service_adder_server:main',
5.          'service_object_client = my_learning_service.service_object_client:main',
6.         ],
7.     },

复制代码

2.2.2 服务端

在my_learning_service文件夹下创建service_object_server.py；

1. """
2. ROS2服务示例-提供目标识别服务
4. @author: zy
5. @since : 2025/12/12
6. """
8. import rclpy                                              # ROS2 Python接口库
9. from rclpy.node import Node                               # ROS2 节点类
10. from sensor_msgs.msg import Image                         # 图像消息类型
11. import numpy as np                                        # Python数值计算库
12. from cv_bridge import CvBridge                            # ROS与OpenCV图像转换类
13. import cv2                                                # Opencv图像处理库
14. from my_learning_interface.srv import GetObjectPosition   # 自定义的服务接口
16. # 橘子的HSV颜色范围
17. lower_orange = np.array([10, 100, 100])     # 橙色的HSV阈值下限
18. upper_orange = np.array([25, 255, 255])     # 橙色的HSV阈值上限
20. class ImageSubscriber(Node):
21.     def __init__(self, name):
22.         super().__init__(name)                              # ROS2节点父类初始化
23.         self.sub = self.create_subscription(
24.             Image, 'image_raw', self.listener_callback, 10) # 创建订阅者对象（消息类型、话题名、订阅者回调函数、队列长度）
25.         self.cv_bridge = CvBridge()                         # 创建一个图像转换对象，用于OpenCV图像与ROS的图像消息的互相转换
27.         self.srv = self.create_service(GetObjectPosition,   # 创建服务器对象（接口类型、服务名、服务器回调函数）
28.                                        'get_target_position',
29.                                        self.object_position_callback)   
30.         self.objectX = 0
31.         self.objectY = 0                              
33.     def object_detect(self, image):
34.         hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)                # 图像从BGR颜色模型转换为HSV模型
35.         mmask_orange = cv2.inRange(hsv_img, lower_orange, upper_orange) # 图像二值化
36.         contours, hierarchy = cv2.findContours(
37.             mask_orange , cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)         # 图像中轮廓检测
39.         for cnt in contours:                                  # 去除一些轮廓面积太小的噪声
40.             if cnt.shape[0] < 150:
41.                 continue
43.             (x, y, w, h) = cv2.boundingRect(cnt)              # 得到橘子所在轮廓的左上角xy像素坐标及轮廓范围的宽和高
44.             cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)# 将橘子的轮廓勾勒出来
45.             cv2.circle(image, (int(x+w/2), int(y+h/2)), 5,
46.                        (0, 255, 0), -1)                       # 将橘子的图像中心点画出来
48.             self.objectX = int(x+w/2)
49.             self.objectY = int(y+h/2)
51.         cv2.imshow("object", image)                            # 使用OpenCV显示处理后的图像效果
52.         cv2.waitKey(50)
54.     def listener_callback(self, data):
55.         self.get_logger().info('Receiving video frame')        # 输出日志信息，提示已进入回调函数
56.         image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')     # 将ROS的图像消息转化成OpenCV图像
57.         self.object_detect(image)                              # 橘子检测
59.     def object_position_callback(self, request, response):     # 创建回调函数，执行收到请求后对数据的处理
60.         if request.get == True:
61.             response.x = self.objectX                          # 目标物体的XY坐标
62.             response.y = self.objectY
63.             self.get_logger().info('Object position\nx: %d y: %d' %
64.                                    (response.x, response.y))   # 输出日志信息，提示已经反馈
65.         else:
66.             response.x = 0
67.             response.y = 0
68.             self.get_logger().info('Invalid command')          # 输出日志信息，提示已经反馈
69.         return response
72. def main(args=None):                                 # ROS2节点主入口main函数
73.     rclpy.init(args=args)                            # ROS2 Python接口初始化
74.     node = ImageSubscriber("service_object_server")  # 创建ROS2节点对象并进行初始化
75.     rclpy.spin(node)                                 # 循环等待ROS2退出
76.     node.destroy_node()                              # 销毁节点对象
77.     rclpy.shutdown()                                 # 关闭ROS2 Python接口

复制代码

完成代码的编写后需要设置功能包的编译选项，让系统知道Python程序的入口，打开功能包的setup.py文件，加入如下入口点的配置：

1.     entry_points={
2.         'console_scripts': [
3.          'service_adder_client  = my_learning_service.service_adder_client:main',
4.          'service_adder_server  = my_learning_service.service_adder_server:main',
5.          'service_object_client = my_learning_service.service_object_client:main',
6.          'service_object_server = my_learning_service.service_object_server:main',
7.         ],
8.     },

复制代码

2.2.3 编译运行

编译程序：

1. pi@NanoPC-T6:~/dev_ws$ colcon build --paths src/my_learning_service

复制代码

启动第一个终端，第一个是视觉识别节点，订阅图像数据，并且集成了一个服务器端对象，随时准备提供目标位置；

1. pi@NanoPC-T6:~/dev_ws$ ros2 run my_learning_service service_object_server

复制代码

启动第二个终端，第二个节点是客户端，当需要根据目标运动时，就发送一次请求，然后拿到一个当前的目标位置；

1. pi@NanoPC-T6:~/dev_ws$ ros2 run my_learning_service service_object_client

复制代码

启动第三个终端，第三个节点是相机驱动节点，发布图像数据；

1. pi@NanoPC-T6:~/dev_ws$  ros2 run usb_cam usb_cam_node_exe

复制代码

2.3 服务命令行操作

查看服务列表:

1. pi@NanoPC-T6:~/dev_ws$ ros2 service list     

复制代码

查看服务数据类型:

1. pi@NanoPC-T6:~/dev_ws$ ros2 service type <service_name>

复制代码

发送服务请求：

1. pi@NanoPC-T6:~/dev_ws$ ros2 service call <service_name> <service_type> <service_data>

复制代码

2.4  思考题

话题和服务是ROS中最为常用的两种数据通信方法；

• 前者适合传感器、控制指令等周期性、单向传输的数据
  
• 后者适合一问一答，同步性要求更高的数据，比如获取机器视觉识别到的目标位置。
  

在机器人开发过程中，类似的通信应用比比皆是，ROS针对绝大部分通用场景，都设计了标准的话题和服务数据类型，比如图像数据、雷达数据、里程计数据等等，不过机器人软硬件繁杂，很多时候这些标准定义也无法满足我们的需求，这个时候，我们就要自定义通信接口了。
参考文章
[1] 古月居ROS2入门教程学习笔记

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！