3、ROS2核心通信机制（下）：服务通信模型、客户端与服务器编程、自定义服务接口、动作通信模型、动作服务器与客户端编程
好，咱们接着聊ROS2的通信机制。上一章我们把话题通信聊透了，这一章我重点讲讲服务和动作。说实话，在我刚接触ROS2那会儿，总觉得话题通信够用了，干嘛还要搞服务和动作？后来在项目里踩了几次坑才明白——不同的通信模式，对应的是不同的应用场景。你想想看，如果所有通信都用话题，那有些场景反而会变得很别扭。
3.1 服务通信模型：一问一答的默契
服务通信，说白了就是客户端发一个请求，服务器回一个响应。这种模式在工程里太常见了。比如你让机器人去抓一个杯子，你问它「抓到了吗」，它回答「抓到了」或者「没抓到」。这就是典型的服务通信。
我个人习惯把服务通信比作「打电话」——你拨号过去，对方接起来，你们完成一次对话，然后挂断。而话题通信更像是「微信群聊」——你发一条消息，所有在群里的人都能看到，但没人保证一定会回复你。
服务通信的核心特点：
- 同步通信：客户端发送请求后会阻塞等待，直到收到响应
- 一对一：一个服务只能被一个客户端调用（但可以有多个客户端同时请求）
- 有反馈：调用方明确知道结果
- 适合短任务：比如查询状态、触发一次性动作
我在项目中遇到过一个问题：用服务通信去控制机械臂走一段复杂的轨迹。结果客户端一直卡在那里等响应，机械臂走了十几秒，客户端就阻塞了十几秒。后来我才意识到，这种长时间的任务根本不适合用服务——你应该用动作通信。
3.2 客户端与服务器编程：手把手写一个加法服务
咱们直接上代码。先写一个服务端，实现两个整数相加的功能。
服务端代码（add_two_ints_server.py）：
import rclpy
from rclpy.node import Node
from example_interfaces.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)
server = AddTwoIntsServer()
rclpy.spin(server)
server.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
客户端代码（add_two_ints_client.py）：
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts
import sys
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
self.future = self.cli.call_async(self.req)
rclpy.spin_until_future_complete(self, self.future)
return self.future.result()
def main(args=None):
rclpy.init(args=args)
client = AddTwoIntsClient()
a = int(sys.argv[1]) if len(sys.argv) > 1 else 3
b = int(sys.argv[2]) if len(sys.argv) > 2 else 5
response = client.send_request(a, b)
client.get_logger().info(f'结果: {a} + {b} = {response.sum}')
client.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
小提示：注意客户端里用了 wait_for_service 来等待服务端上线。我曾经在调试时忘了加这个，结果客户端一启动就报错——服务端还没跑起来呢。这个等待机制是必须的，别偷懒。
3.3 自定义服务接口：打造你自己的通信协议
用现成的 AddTwoInts 当然方便，但实际项目中你肯定需要自定义接口。比如你要控制机器人抓取物体，请求里需要包含目标位置、抓取力度、超时时间等信息。
自定义服务接口的步骤很简单：
- 在功能包下创建
srv文件夹 - 编写
.srv文件 - 修改
CMakeLists.txt和package.xml - 编译生成代码
举个例子，创建一个抓取服务的接口：
# GraspObject.srv
# 请求部分
string object_name # 目标物体名称
float32 grasp_force # 抓取力度（牛顿）
float32 timeout # 超时时间（秒）
---
# 响应部分
bool success # 是否成功
string message # 结果描述
float32 actual_force # 实际施加的力度
这里有个细节要注意：--- 上面是请求，下面是响应。中间用三个短横线隔开。我刚开始写的时候老忘记这个分隔符，编译一直报错，查了半天才发现是格式问题。
避坑指南：我曾经在自定义接口里用了 int 类型，结果编译报错说找不到。ROS2的接口类型必须用 int32、float64 这种带位数的写法。别问我怎么知道的——翻了好久的文档才找到原因。
3.4 动作通信模型：给任务一个「进度条」
好，现在聊聊动作通信。为什么需要它？你想想看，如果你让机器人从A点走到B点，这个过程可能需要几十秒甚至几分钟。用话题通信？你得自己维护状态机。用服务通信？客户端会一直阻塞。这时候动作通信就派上用场了。
动作通信的核心是三个部分：
- 目标（Goal）：你要让机器人做什么
- 反馈（Feedback）：机器人做到哪一步了
- 结果（Result）：最终做完了没有
我个人觉得，动作通信最妙的地方在于它允许你「中途取消」。比如你让机器人去拿杯子，走到一半你发现杯子其实在另一个地方，你可以直接取消当前动作，重新发一个目标。这在话题和服务里实现起来都很麻烦。
3.5 动作服务器与客户端编程：让机器人走个正方形
咱们写一个动作服务器，控制机器人走一个正方形的轨迹。这里用 nav2_msgs/action/NavigateToPose 作为例子，但为了演示，我们自己定义一个简单的动作接口。
先定义动作接口（SquarePath.action）：
# 目标
float32 side_length # 正方形边长（米）
---
# 结果
bool success
float32 total_distance # 总行走距离
---
# 反馈
float32 current_x # 当前位置x
float32 current_y # 当前位置y
float32 progress # 进度百分比（0-100）
动作服务器端（简化版）：
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from your_package.action import SquarePath
class SquarePathServer(Node):
def __init__(self):
super().__init__('square_path_server')
self._action_server = ActionServer(
self,
SquarePath,
'square_path',
self.execute_callback)
async def execute_callback(self, goal_handle):
self.get_logger().info(f'收到目标: 边长 {goal_handle.request.side_length}米')
feedback_msg = SquarePath.Feedback()
# 模拟走正方形的四个边
for i in range(4):
# 模拟移动
for step in range(25):
feedback_msg.current_x = step * 0.04
feedback_msg.current_y = 0.0
feedback_msg.progress = (i * 25 + step)
goal_handle.publish_feedback(feedback_msg)
await asyncio.sleep(0.1)
# 检查是否被取消
if goal_handle.is_cancel_requested:
goal_handle.canceled()
return SquarePath.Result(success=False, total_distance=0.0)
goal_handle.succeed()
result = SquarePath.Result()
result.success = True
result.total_distance = goal_handle.request.side_length * 4
return result
动作客户端：
import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from your_package.action import SquarePath
class SquarePathClient(Node):
def __init__(self):
super().__init__('square_path_client')
self._action_client = ActionClient(self, SquarePath, 'square_path')
def send_goal(self, side_length):
goal_msg = SquarePath.Goal()
goal_msg.side_length = side_length
self._action_client.wait_for_server()
self._send_goal_future = self._action_client.send_goal_async(
goal_msg,
feedback_callback=self.feedback_callback)
self._send_goal_future.add_done_callback(self.goal_response_callback)
def feedback_callback(self, feedback_msg):
fb = feedback_msg.feedback
self.get_logger().info(f'进度: {fb.progress}%, 位置: ({fb.current_x:.2f}, {fb.current_y:.2f})')
def goal_response_callback(self, future):
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().info('目标被拒绝')
return
self.get_logger().info('目标被接受')
self._get_result_future = goal_handle.get_result_async()
self._get_result_future.add_done_callback(self.get_result_callback)
def get_result_callback(self, future):
result = future.result().result
self.get_logger().info(f'结果: 成功={result.success}, 总距离={result.total_distance:.2f}米')
关键点总结：
- 动作服务器用
execute_callback处理目标，可以异步执行 - 客户端通过
feedback_callback实时获取进度 - 支持取消操作，服务器端要检查
is_cancel_requested - 动作通信适合长时间任务，比如导航、机械臂运动
嗯，这里要注意一个细节：动作通信的反馈是「可选的」。也就是说，你可以不发反馈，只返回最终结果。但我在实际项目中强烈建议你加上反馈——调试的时候太有用了。有一次机器人导航卡在半路，我一看反馈信息，发现进度停在37%不动了，立刻就知道是路径规划出了问题。
好了，这一章的内容就到这里。服务通信和动作通信是ROS2里非常重要的两个机制，尤其是动作通信，在机器人运动控制中几乎是标配。下一章我们聊聊参数服务器和生命周期节点，这两个东西在工程里也经常用到。