4、ROS2通信机制实战（二）：使用Python编写服务端与客户端节点、自定义服务类型（srv）、同步与异步调用、使用命令行工具（ros2 service）调试。

上一章我们聊了话题通信，那是ROS2里最常用的“发布-订阅”模式。但说实话，很多场景下你需要的不是单向的数据流，而是“请求-应答”模式。比如你让机器人去抓一个杯子，你得等它告诉你“抓到了”或者“没抓到”。这时候，服务（Service）通信就该登场了。

我个人习惯把服务通信比作打电话。你拨号（发起请求），对方接听并处理，然后给你一个答复。而话题通信更像是听收音机，你只管听，没人会专门回复你。好，咱们今天就把这个“电话系统”给搭起来。

4.1 服务通信的核心概念

服务通信基于客户端-服务器模型。一个服务端节点提供某个功能，客户端节点发起请求并等待响应。整个过程是同步的，但ROS2也支持异步调用，我们后面会细说。

服务的核心是服务类型（srv文件）。它定义了请求和响应的数据结构。你想想看，没有这个约定，客户端和服务端怎么知道该发什么、回什么？

4.2 自定义服务类型（srv）

ROS2内置了一些标准服务类型，比如std_srvs/srv/Empty。但实际项目中，我们几乎总是需要自定义。就拿“加法器”这个经典例子来说吧。

首先，在功能包目录下创建srv文件夹，然后新建AddTwoInts.srv文件：

int64 a
int64 b
---
int64 sum

这里---上面是请求结构，下面是响应结构。简单明了。

别忘了在CMakeLists.txt里添加依赖：

find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "srv/AddTwoInts.srv"
)

然后编译：colcon build --packages-select 你的包名。编译成功后，你就可以在代码里导入这个服务类型了。

4.3 编写服务端节点

服务端要做的很简单：创建一个服务，然后等待请求，处理，返回结果。来看代码：

import rclpy
from rclpy.node import Node
from your_package.srv import AddTwoInts

class AddTwoIntsServer(Node):
    def __init__(self):
        super().__init__('add_two_ints_server')
        self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_callback)
        self.get_logger().info('服务端已启动，等待请求...')

    def add_callback(self, request, response):
        response.sum = request.a + request.b
        self.get_logger().info(f'收到请求: {request.a} + {request.b} = {response.sum}')
        return response

def main(args=None):
    rclpy.init(args=args)
    node = AddTwoIntsServer()
    rclpy.spin(node)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

这里有个细节：回调函数add_callback必须返回response对象。我曾经见过有人忘记写return，结果客户端一直卡在那里等响应。嗯，这种bug排查起来还挺隐蔽的。

4.4 编写客户端节点

客户端稍微复杂一点，因为它涉及同步和异步两种调用方式。我们先看同步调用：

import rclpy
from rclpy.node import Node
from your_package.srv import AddTwoInts

class AddTwoIntsClient(Node):
    def __init__(self):
        super().__init__('add_two_ints_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().warn('等待服务端上线...')
        self.req = AddTwoInts.Request()

    def send_request(self, a, b):
        self.req.a = a
        self.req.b = b
        # 同步调用
        future = self.cli.call_async(self.req)
        rclpy.spin_until_future_complete(self, future)
        return future.result()

def main(args=None):
    rclpy.init(args=args)
    node = AddTwoIntsClient()
    response = node.send_request(3, 5)
    node.get_logger().info(f'结果: {response.sum}')
    node.destroy_node()
    rclpy.shutdown()

你可能会问：为什么call_async是异步方法，却用在同步场景里？其实spin_until_future_complete这个函数会阻塞当前线程，直到future完成。说白了，这就是用异步接口实现了同步效果。

4.4.1 真正的异步调用

如果你不想阻塞主线程，可以用回调函数处理响应：

def send_request_async(self, a, b):
    self.req.a = a
    self.req.b = b
    future = self.cli.call_async(self.req)
    future.add_done_callback(self.response_callback)

def response_callback(self, future):
    response = future.result()
    self.get_logger().info(f'异步收到结果: {response.sum}')

我个人更推荐异步方式，尤其是在需要同时处理多个服务调用或话题订阅的场景下。同步调用虽然简单，但会阻塞节点，影响实时性。

4.5 使用命令行工具调试

ROS2的命令行工具是调试利器。启动服务端后，你可以用以下命令查看服务列表：

ros2 service list

输出应该包含/add_two_ints。想看服务类型？

ros2 service type /add_two_ints

这会返回your_package/srv/AddTwoInts。如果你想直接调用服务测试：

ros2 service call /add_two_ints your_package/srv/AddTwoInts "{a: 10, b: 20}"

终端会立刻返回响应结果。这个命令在调试时特别有用，不用写客户端代码就能验证服务端逻辑。

4.6 实战中的注意事项

场景	推荐方式	原因
快速原型验证	同步调用	代码简单，易于理解
多任务并发	异步调用	不阻塞主循环
服务端可能延迟	异步调用 + 超时处理	避免永久阻塞
命令行快速测试	ros2 service call	无需编写代码

最后提醒一点：服务通信的可靠性依赖于网络和服务端状态。如果服务端崩溃，客户端调用会抛出异常。记得加try-except处理，别让程序直接挂掉。

好了，服务通信的核心内容就这些。动手写一个加法器服务，再用命令行调一调，你就能彻底掌握它了。