4. 服务通信实战：同步/异步客户端、服务端多线程处理、超时与重试机制

服务通信，说白了就是ROS2里的“一问一答”模式。你发一个请求，它回一个响应。听起来简单，但实际用起来坑不少。

我记得刚带团队做第一个机器人项目时，就栽在服务通信上。客户端发了个请求，服务端处理太慢，客户端直接卡死。整个系统像死了一样。从那以后，我对服务通信的设计就格外小心。

今天咱们就把这块彻底讲透。同步客户端、异步客户端、服务端多线程、超时重试，一个一个来。

4.1 同步客户端：简单但危险

同步客户端，就是发请求后一直等，等到响应回来才继续往下走。

代码写起来很直观：

# 同步客户端示例
from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node

class SyncClient(Node):
    def __init__(self):
        super().__init__('sync_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        
        # 等待服务端上线
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('服务未就绪，继续等待...')
        
        self.req = AddTwoInts.Request()
        
    def call_service(self, a, b):
        self.req.a = a
        self.req.b = b
        
        # 同步调用，会阻塞
        self.future = self.cli.call_async(self.req)
        rclpy.spin_until_future_complete(self, self.future)
        
        if self.future.result() is not None:
            return self.future.result().sum
        else:
            self.get_logger().error('服务调用失败')
            return None

看到没？spin_until_future_complete 这行就是阻塞点。在等待期间，节点啥也干不了。

4.2 异步客户端：真正的实战选择

异步客户端就不一样了。发完请求，该干嘛干嘛。响应回来了，回调函数自动处理。

# 异步客户端示例
class AsyncClient(Node):
    def __init__(self):
        super().__init__('async_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('等待服务...')
        
        self.req = AddTwoInts.Request()
        
    def send_request(self, a, b):
        self.req.a = a
        self.req.b = b
        
        # 异步调用，不阻塞
        self.future = self.cli.call_async(self.req)
        
        # 添加回调，响应回来后自动处理
        self.future.add_done_callback(self.response_callback)
        
    def response_callback(self, future):
        try:
            response = future.result()
            self.get_logger().info(f'收到响应: {response.sum}')
        except Exception as e:
            self.get_logger().error(f'服务调用异常: {e}')

你看，call_async 调用后立即返回。节点可以继续处理其他事情。响应回来了，response_callback 自动触发。

4.3 服务端多线程处理

服务端默认是单线程的。什么意思？就是一次只能处理一个请求。如果第一个请求处理了5秒，第二个请求就得排队等5秒。

你想想看，如果多个客户端同时发请求，单线程服务端就成了瓶颈。

解决办法？用多线程回调组。

# 多线程服务端示例
from rclpy.executors import MultiThreadedExecutor
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup

class MultiThreadServer(Node):
    def __init__(self):
        super().__init__('multi_thread_server')
        
        # 创建独立的回调组
        self.cb_group = MutuallyExclusiveCallbackGroup()
        
        # 将服务绑定到该回调组
        self.srv = self.create_service(
            AddTwoInts, 
            'add_two_ints', 
            self.handle_request,
            callback_group=self.cb_group
        )
        
    def handle_request(self, request, response):
        # 模拟耗时操作
        import time
        time.sleep(2)
        
        response.sum = request.a + request.b
        self.get_logger().info(f'处理完成: {request.a} + {request.b} = {response.sum}')
        return response

# 使用多线程执行器
def main():
    rclpy.init()
    node = MultiThreadServer()
    
    # 创建多线程执行器，指定线程数
    executor = MultiThreadedExecutor(num_threads=4)
    executor.add_node(node)
    
    try:
        executor.spin()
    finally:
        executor.shutdown()
        node.destroy_node()
        rclpy.shutdown()

这里的关键点有两个：

• 回调组：MutuallyExclusiveCallbackGroup 表示这个组里的回调是互斥的，但不同组之间可以并行
• 多线程执行器：MultiThreadedExecutor 负责分配线程

4.4 超时与重试机制

网络通信，不可靠是常态。服务端可能挂了，可能网络延迟，可能处理超时。没有超时重试机制，系统就是纸糊的。

ROS2的服务客户端本身没有内置超时参数。需要我们自己实现。

# 带超时重试的客户端
import asyncio

class RobustClient(Node):
    def __init__(self):
        super().__init__('robust_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        self.req = AddTwoInts.Request()
        
        # 配置参数
        self.max_retries = 3
        self.timeout_seconds = 2.0
        self.retry_delay = 0.5
        
    async def call_with_retry(self, a, b):
        self.req.a = a
        self.req.b = b
        
        for attempt in range(1, self.max_retries + 1):
            self.get_logger().info(f'第 {attempt} 次尝试...')
            
            # 检查服务是否可用
            if not self.cli.service_is_ready():
                self.get_logger().warn('服务不可用，等待...')
                await asyncio.sleep(self.retry_delay)
                continue
            
            # 发送异步请求
            future = self.cli.call_async(self.req)
            
            try:
                # 等待指定时间
                result = await asyncio.wait_for(
                    asyncio.wrap_future(future), 
                    timeout=self.timeout_seconds
                )
                return result.sum
                
            except asyncio.TimeoutError:
                self.get_logger().warn(f'第 {attempt} 次超时')
                # 取消未完成的future
                future.cancel()
                
                if attempt < self.max_retries:
                    await asyncio.sleep(self.retry_delay)
                continue
                
            except Exception as e:
                self.get_logger().error(f'异常: {e}')
                break
        
        self.get_logger().error('所有重试均失败')
        return None

这个实现里，我用了 asyncio.wait_for 来实现超时。为什么不用 time.sleep？因为 sleep 会阻塞整个线程，而 asyncio.sleep 不会。

4.5 实战经验总结

说了这么多，我总结几条实战经验：

嗯，服务通信这块就讲到这里。下一章咱们聊话题通信的进阶用法，包括QoS配置和零拷贝传输。到时候见。