本文主要结合简单示例谈一下自己对ros2学习的一下体会。1.服务接口1.服务接口是什么1.我的浅显理解是服务接口是定义服务的客户端和服务端之间的数据传输类型。拿海龟模拟器举例子启动海龟模拟器后 ros2 run turtlesim turtlesim_node查看服务类型[]前的为服务名称后面为接口类型拿/spawn为例/spawn为服务名称[turtlesim/srv/Spawn]中的Spawn为该接口的接口类型查看该接口格式ros2 interface show turtlesim/srv/Spawn---上部分为请求参数下为响应参数尝试调用该服务ros2 service call /spawn turtlesim/srv/Spawn {x: 1, y: 1}参数1为服务名参数2服务类型参数3请求参数 该服务作用是在(1,1)位置产生新海龟2.如何自定义一个服务接口自定义服务接口就是创建一个新的.srv文件它定义了你这个服务的“请求”和“响应”的数据格式。1.创建功能包拿tutorial_interfaces为例子ros2 pkg create --build-type ament_cmake tutorial_interfaces2.创建自定义服务接口文件 (.srv)cd tutorial_interfacesmkdir srv在srv文件夹中创建一个名为AddThreeInts.srv的文件-。其内容定义了服务的请求和响应结构请求和响应之间用---分隔int64 aint64 bint64 c---int64 sum3.修改CMakeLists.txt编辑tutorial_interfaces目录下的CMakeLists.txt文件完成两件事1.找到rosidl_default_generators 修改为rosidl_generate_interfaces(${PROJECT_NAME}srv/AddThreeInts.srv)srv/AddThreeInts.srv就是刚才创建的接口文件如果该服务接口使用到了别的依赖也需要加入比如rosidl_generate_interfaces(${PROJECT_NAME}srv/AddThreeInts.srv,DEPENDENCIES sensor_msgs)4.修改package.xml添加声明声明该功能包为一个消息接口功能包member_of_grouprosidl_interface_packages/member_of_group5.编译功能包colcon build --packages-select tutorial_interfaces6.验证与使用验证编译成功后可以用以下命令检查接口是否被正确识别在其他节点中使用在需要使用这个自定义服务的C或Python节点中像导入标准接口一样导入它即可from tutorial_interfaces.srv import AddThreeInts2.服务1.什么是ROS2服务ROS 2的服务是一种同步、双向、短时的通信机制采用客户端/服务器模型。它非常适合那些需要即时反馈和快速计算的场景例如查询节点状态或触发一个瞬时动作。它基于客户端-服务器Client-Server模型。客户端发送请求给服务端服务端获得请求后经过处理将结果返回客户端。过程是一问一答一次性的。在ROS 2中每个服务名Service Name下只能有一个服务服务器Server但可以有任意多个客户端Client调用它。2.如何自定义一个服务设计一个服务1.客户端#!/usr/bin/env python3 import rclpy from rclpy.node import Node from turtlesim.srv import Spawn import sys class TurtleClient(Node): def __init__(self, node_nameturtle_client): super().__init__(node_name) self.client self.create_client(Spawn, /turtle_spawn) while not self.client.wait_for_service(timeout_sec1.0): self.get_logger().warn(等待 /turtle_spawn 服务上线...) def send_spawn_request(self, x, y, theta, name): req Spawn.Request() req.x x req.y y req.theta theta req.name name future self.client.call_async(req) rclpy.spin_until_future_complete(self, future) if future.result() is not None: self.get_logger().info(f服务器返回乌龟名称: {future.result().name}) return future.result().name else: self.get_logger().error(请求失败) return None def main(argsNone): rclpy.init(argsargs) node TurtleClient() # 从命令行参数读取坐标可选 if len(sys.argv) 4: x float(sys.argv[1]) y float(sys.argv[2]) theta float(sys.argv[3]) name sys.argv[4] if len(sys.argv) 4 else else: x, y, theta, name 5.0, 5.0, 0.0, turtle2 node.send_spawn_request(x, y, theta, name) node.destroy_node() rclpy.shutdown() if __name__ __main__: main()2.服务端#!/usr/bin/env python3 import rclpy from rclpy.node import Node from turtlesim.srv import Spawn from rclpy.callback_groups import MutuallyExclusiveCallbackGroup from rclpy.executors import MultiThreadedExecutor class TurtleServer(Node): def __init__(self, node_nameturtle_server): super().__init__(node_name) # 创建客户端用于连接 turtlesim 的 /spawn 服务 self.spawn_client self.create_client(Spawn, /spawn) # 等待 /spawn 服务可用超时 5 秒 while not self.spawn_client.wait_for_service(timeout_sec1.0): self.get_logger().warn(等待 /spawn 服务上线...) # 创建自己的服务接收来自客户端的请求 self.srv self.create_service( Spawn, /turtle_spawn, self.spawn_callback, callback_groupMutuallyExclusiveCallbackGroup() # 保证线程安全 ) self.get_logger().info(turtle_server 已启动监听 /turtle_spawn) def spawn_callback(self, request, response): 服务回调接收客户端的 Spawn 请求转发给 /spawn并返回结果。 self.get_logger().info( f收到生成请求: x{request.x}, y{request.y}, theta{request.theta}, name{request.name} ) # 构造转发请求 forward_req Spawn.Request() forward_req.x request.x forward_req.y request.y forward_req.theta request.theta forward_req.name request.name # 同步调用 /spawn 服务阻塞直至收到响应或超时 future self.spawn_client.call_async(forward_req) rclpy.spin_until_future_complete(self, future, timeout_sec5.0) if future.result() is not None: response.name future.result().name self.get_logger().info(f成功生成乌龟: {response.name}) else: # 若调用失败设置空名称并记录错误 response.name self.get_logger().error(转发 /spawn 请求失败请检查 turtlesim 是否运行) return response def main(argsNone): rclpy.init(argsargs) node TurtleServer() # 使用多线程执行器以便同时处理多个请求如果需要 executor MultiThreadedExecutor() executor.add_node(node) try: executor.spin() except KeyboardInterrupt: pass finally: node.destroy_node() rclpy.shutdown() if __name__ __main__: main()3.服务端模板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.serviceself.create_service(AddTwoInts,/add_two_ints,self.callback) self.get_logger().info(加法服务端启动) def callback(self,request,response): response.sumrequest.arequest.b self.get_logger().info(f处理结果{request.a}{request.b}{response.sum}) return response def main(): rclpy.init() nodeAddTwoIntsServer() rclpy.spin(node) node.destroy_node() rclpy.shutdown()该接口类型定义为dysdys-virtual-machine:~$ ros2 interface show example_interfaces/srv/AddTwoIntsint64 aint64 b---int64 sumdysdys-virtual-machine:~$4.客户端模板import rclpy from rclpy.node import Node from example_interfaces.srv import AddTwoInts class AddTwoIntsClient(Node): def __init__(self): super().__init__(add_two_ints_client) self.client self.create_client(AddTwoInts, /add_two_ints) while not self.client.wait_for_service(timeout_sec1.0): self.get_logger().info(服务端未启动) def send_request(self): req AddTwoInts.Request() req.a 5 req.b 3 self.get_logger().info(发送请求...) # 异步调用并绑定回调 future self.client.call_async(req) future.add_done_callback(self.response_callback) def response_callback(self, future): response future.result() self.get_logger().info(f获得服务端结果{response.sum}) def main(): rclpy.init() node AddTwoIntsClient() node.send_request() rclpy.spin(node) node.destroy_node() rclpy.shutdown()核心写法 异步加回调阻塞等待型写法import rclpy from rclpy.node import Node from example_interfaces.srv import AddTwoInts class AddTwoIntsClient(Node): def __init__(self): super().__init__(add_two_ints_client) self.client self.create_client(AddTwoInts, /add_two_ints) while not self.client.wait_for_service(timeout_sec1.0): self.get_logger().info(等待服务端...) def send_request(self): req AddTwoInts.Request() req.a 5 req.b 3 self.get_logger().info(发送请求...) future self.client.call_async(req) # 等待完成超时5秒 rclpy.spin_until_future_complete(self, future, timeout_sec5.0) if future.done(): response future.result() self.get_logger().info(f结果{response.sum}) else: self.get_logger().error(请求超时) def main(): rclpy.init() node AddTwoIntsClient() node.send_request() node.destroy_node() rclpy.shutdown() if __name__ __main__: main()异步回调写法import rclpy from rclpy.node import Node from example_interfaces.srv import AddTwoInts class AddTwoIntsClient(Node): def __init__(self): super().__init__(add_two_ints_client) self.client self.create_client(AddTwoInts, /add_two_ints) while not self.client.wait_for_service(timeout_sec1.0): self.get_logger().info(服务端未启动) def send_request(self): req AddTwoInts.Request() req.a 5 req.b 3 self.get_logger().info(发送请求...) # 异步调用并绑定回调 future self.client.call_async(req) future.add_done_callback(self.response_callback) def response_callback(self, future): response future.result() self.get_logger().info(f获得服务端结果{response.sum}) def main(): rclpy.init() node AddTwoIntsClient() node.send_request() rclpy.spin(node) node.destroy_node() rclpy.shutdown()5.区别 核心区别一览对比维度add_done_callback异步回调spin_until_future_complete同步等待是否阻塞不阻塞代码瞬间返回阻塞卡住直到拿到结果或超时结果获取方式在回调函数里被动接收直接return future.result()主动拿到对spin的依赖必须在spin循环中才能触发回调不依赖函数内部自己处理事件节点可处理其他任务可以多任务并发不可以死等一个结果代码写法复杂回调函数 绑定简单像普通函数调用一样直观适用场景节点持续运行需同时处理多件事一次性调用或启动时的初始化配置 实战中怎么选✅ 用异步回调的场景节点需要长期运行如机器人控制节点。节点同时要处理多个任务如订阅雷达数据 调用服务 响应参数修改。服务响应时间不确定可能很长或可能超时。你需要并发发送多个服务请求让它们各自回调互不干扰。典型代码模式rclpy.spin(node) # 节点常驻等待各种事件✅ 用同步等待的场景节点启动时做一次性配置如从参数服务器拉取初始参数。脚本型程序调用完服务就退出。你的逻辑是顺序执行的后续代码依赖服务结果。不希望写回调函数追求代码简洁。典型代码模式result node.call_some_service() # 阻塞 # 拿到结果后做后续处理 process_result(result)3.话题1.什么是话题ROS 2话题Topic是节点间进行异步数据通信的发布/订阅Publish-Subscribe模型的核心2.发布者发布者模板#!/usr/bin/env python3 import rclpy from rclpy.node import Node from std_msgs.msg import String class PublisherNode(Node): def __init__(self): super().__init__(publisher_node) self.publisher_ self.create_publisher(String, chatter, 10) self.timer_ self.create_timer(1.0, self.publish_callback) self.counter 0 def publish_callback(self): msg String() msg.data fHello ROS2! {self.counter} self.publisher_.publish(msg) self.get_logger().info(fPublishing: {msg.data}) self.counter 1 def main(argsNone): rclpy.init(argsargs) node PublisherNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ __main__: main()在构造函数内实例发布者对象time内调用回调函数回调函数内放入发布信息3.订阅者#!/usr/bin/env python3 import rclpy from rclpy.node import Node from std_msgs.msg import String class SubscriberNode(Node): def __init__(self): super().__init__(subscriber_node) # 创建订阅者主题为 chatter队列深度 10回调函数为 callback self.subscription_ self.create_subscription( String, chatter, self.callback, 10 ) self.get_logger().info(Subscriber started, waiting for messages...) def callback(self, msg): # 收到消息时执行 self.get_logger().info(fReceived: {msg.data}) def main(argsNone): rclpy.init(argsargs) node SubscriberNode() rclpy.spin(node) # 阻塞等待消息 node.destroy_node() rclpy.shutdown() if __name__ __main__: main()实例化订阅者对象参数为回调函数回调函数为接收信息后的处理方法。实例import rclpy from rclpy.node import Node from geometry_msgs.msg import Twist from turtlesim.msg import Pose class turtle(Node): def __init__(self,node_name): super().__init__(node_name) # 定义速度 self.linear1.0 self.angular0.5 # 发布者 话题类型, self.publisherself.create_publisher(Twist,/turtle1/cmd_vel, 10) # 定时器 self.timerself.create_timer(0.1,self.callback) # 订阅者 self.subscriptionself.create_subscription(Pose,/turtle1/pose, self.pose_callback, 10) self.get_logger().info(CircleTurtleNode 已启动乌龟开始画圆...) # 定时器回调函数 def callback(self): # 定义速度 twistTwist() twist.linear.xself.linear twist.angular.zself.angular # 发布 self.publisher.publish(twist) #订阅回调函数 def pose_callback(self,msg:Pose): linearmsg.linear_velocity angularmsg.angular_velocity self.get_logger().info(f线速度:{linear},角速度:{angular}) def main(): # 初始化节点 rclpy.init() nodeturtle(turtle_node) rclpy.spin(node) rclpy.shutdown()