4. 服务通信实战：客户端与服务器模型，同步/异步调用
服务通信，说白了就是ROS2里一种「一问一答」的机制。一个节点发出请求，另一个节点处理并返回结果。这跟话题通信那种「一直广播、谁爱听谁听」的模式完全不同。
我个人习惯把服务通信比作「打电话」。你拨号（发请求），对方接听并回答（返回响应），然后挂断。而话题通信更像是「听收音机」——你打开就能听，但没法跟电台对话。
核心要点：服务通信适用于需要即时反馈的场景，比如控制机器人抓取物体、查询传感器状态、触发某个动作。
4.1 服务通信的底层逻辑
先看一张图，帮你快速建立整体认知：
嗯，这张图其实已经说清楚了。客户端发请求，服务端处理完返回结果。整个过程是同步的——客户端发完请求后，会一直等着服务端回复。
4.2 服务接口的定义
在ROS2里，服务接口用 .srv 文件定义。我刚开始学的时候，总把它跟 .msg 搞混。其实区别很简单：.srv 文件里用 --- 分隔了请求和响应两部分。
举个例子，Nav2里常用的 NavigateToPose.srv：
# 请求部分
geometry_msgs/PoseStamped pose
bool use_start_pose
geometry_msgs/PoseStamped start_pose
---
# 响应部分
std_msgs/Empty result你看，--- 上面是客户端要发的内容，下面是服务端返回的内容。这个结构非常清晰。
我的经验：写服务接口时，请求字段别太多。我曾经在一个项目里塞了十几个字段，结果调试时自己都记不清顺序。保持简洁，3-5个字段最合适。
4.3 服务端实现
服务端的任务很简单：注册一个服务，然后等着被调用。来看一个实际例子：
# server.py
import rclpy
from rclpy.node import Node
from nav2_msgs.srv import NavigateToPose
class Nav2Server(Node):
def __init__(self):
super().__init__('nav2_server')
self.srv = self.create_service(
NavigateToPose, # 服务类型
'/navigate_to_pose', # 服务名称
self.handle_request # 回调函数
)
self.get_logger().info('服务端已启动，等待请求...')
def handle_request(self, request, response):
self.get_logger().info(f'收到导航请求：{request.pose}')
# 这里执行实际的导航逻辑
# 比如调用路径规划、控制底盘等
self.get_logger().info('导航完成')
return response
def main(args=None):
rclpy.init(args=args)
server = Nav2Server()
rclpy.spin(server)
server.destroy_node()
rclpy.shutdown()这里要注意的是 create_service 的三个参数。我见过不少新手把服务类型和服务名称搞反。服务类型是 NavigateToPose 这个类，服务名称是字符串 '/navigate_to_pose'。
4.4 客户端实现：同步调用
同步调用是最直接的方式。客户端发请求，然后阻塞等待结果。说白了就是「你不回我，我就不走」。
# client_sync.py
import rclpy
from rclpy.node import Node
from nav2_msgs.srv import NavigateToPose
class Nav2ClientSync(Node):
def __init__(self):
super().__init__('nav2_client_sync')
self.cli = self.create_client(
NavigateToPose,
'/navigate_to_pose'
)
# 等待服务端就绪
while not self.cli.wait_for_service(timeout_sec=1.0):
self.get_logger().warn('等待服务端启动...')
self.get_logger().info('服务端已就绪')
def send_request(self):
req = NavigateToPose.Request()
req.pose.header.frame_id = 'map'
req.pose.pose.position.x = 2.0
req.pose.pose.position.y = 1.0
# 同步调用，会阻塞
future = self.cli.call_async(req)
rclpy.spin_until_future_complete(self, future)
if future.result() is not None:
self.get_logger().info('导航请求成功')
return future.result()
else:
self.get_logger().error('导航请求失败')
return None嗯，这里有个坑。虽然叫 call_async，但配合 spin_until_future_complete 用，效果就是同步的。为什么会这样？因为 spin_until_future_complete 会一直阻塞，直到 future 完成。
我曾经踩过的坑：在回调函数里直接调用同步服务。这会导致死锁！因为回调函数本身就在 spin 线程里，你再 spin 一次就卡死了。解决方案是用异步调用，或者把服务调用放到另一个线程。
4.5 客户端实现：异步调用
异步调用更适合在回调函数或循环中使用。它不会阻塞主线程，而是通过回调来获取结果。
# client_async.py
import rclpy
from rclpy.node import Node
from nav2_msgs.srv import NavigateToPose
class Nav2ClientAsync(Node):
def __init__(self):
super().__init__('nav2_client_async')
self.cli = self.create_client(
NavigateToPose,
'/navigate_to_pose'
)
while not self.cli.wait_for_service(timeout_sec=1.0):
self.get_logger().warn('等待服务端启动...')
self.timer = self.create_timer(5.0, self.timer_callback)
def timer_callback(self):
req = NavigateToPose.Request()
req.pose.header.frame_id = 'map'
req.pose.pose.position.x = 3.0
req.pose.pose.position.y = 2.0
# 异步调用，不阻塞
future = self.cli.call_async(req)
future.add_done_callback(self.response_callback)
self.get_logger().info('请求已发送，继续执行其他任务')
def response_callback(self, future):
try:
response = future.result()
self.get_logger().info('收到服务端响应')
except Exception as e:
self.get_logger().error(f'服务调用异常：{e}')你看，异步调用的核心是 add_done_callback。请求发出去后，程序继续往下走。等服务端返回结果时，回调函数会被自动触发。
4.6 同步 vs 异步：怎么选？
我整理了一个对比表，帮你快速决策：
| 对比维度 | 同步调用 | 异步调用 |
|---|---|---|
| 代码复杂度 | 简单，线性逻辑 | 稍复杂，需要回调 |
| 阻塞行为 | 阻塞等待 | 非阻塞 |
| 适用场景 | 初始化、单次请求 | 循环、回调、高并发 |
| 调试难度 | 容易，堆栈清晰 | 较难，回调链复杂 |
| 性能影响 | 可能阻塞主循环 | 不影响主循环 |
我个人习惯是：初始化阶段用同步，运行阶段用异步。比如机器人启动时，用同步调用确认各个服务是否就绪。正常运行时，用异步调用发送导航指令，不影响其他任务。
4.7 实战：Nav2中的服务调用
在Nav2里，最常用的服务就是 /navigate_to_pose。我拿一个实际项目中的代码片段来说明：
class RobotController(Node):
def __init__(self):
super().__init__('robot_controller')
self.nav_client = self.create_client(
NavigateToPose,
'/navigate_to_pose'
)
# 等待Nav2就绪
self.get_logger().info('等待Nav2服务就绪...')
self.nav_client.wait_for_service()
self.get_logger().info('Nav2已就绪')
def go_to_pose(self, x, y, theta):
"""发送导航目标点"""
req = NavigateToPose.Request()
req.pose.header.stamp = self.get_clock().now().to_msg()
req.pose.header.frame_id = 'map'
req.pose.pose.position.x = x
req.pose.pose.position.y = y
# 使用四元数表示朝向
from tf_transformations import quaternion_from_euler
q = quaternion_from_euler(0, 0, theta)
req.pose.pose.orientation.x = q[0]
req.pose.pose.orientation.y = q[1]
req.pose.pose.orientation.z = q[2]
req.pose.pose.orientation.w = q[3]
# 异步调用
future = self.nav_client.call_async(req)
future.add_done_callback(self.nav_done_callback)
self.get_logger().info(f'发送导航目标：({x:.2f}, {y:.2f}, {theta:.2f})')
def nav_done_callback(self, future):
try:
result = future.result()
self.get_logger().info('导航完成')
except Exception as e:
self.get_logger().error(f'导航失败：{e}')避坑指南：我曾经在发送导航请求时忘了设置时间戳 header.stamp，结果Nav2一直报错说「目标点过期」。后来才发现，Nav2会检查时间戳，如果跟当前时间差太多，就认为目标点无效。所以记得用 self.get_clock().now().to_msg() 设置当前时间。
4.8 调试技巧
服务通信出问题时，别慌。我一般按这个顺序排查：
- 检查服务是否注册：用
ros2 service list查看所有服务 - 检查服务类型：用
ros2 service type /service_name确认类型匹配 - 手动调用测试：用
ros2 service call /service_name type "{...}"测试 - 查看日志：设置
RCLCPP_LOG_LEVEL=DEBUG看详细输出
嗯，说到手动调用，这个命令特别实用。比如测试导航服务：
ros2 service call /navigate_to_pose nav2_msgs/srv/NavigateToPose "{pose: {header: {frame_id: 'map'}, pose: {position: {x: 1.0, y: 2.0}}}}"如果这个命令能正常工作，说明服务端没问题。问题大概率出在客户端代码上。
服务通信这块，说白了就是「请求-响应」模型。同步调用简单直接，异步调用灵活高效。你在实际项目中，根据场景选择合适的模式就好。记住我前面说的那个死锁的坑，还有时间戳的问题，基本就能顺利跑通了。