ROS2使用Python创建服务提供者、消费者

news2025/10/31 10:22:08

1.创建服务提供者

ros2 pkg create example_service_rclpy --build-type ament_python --dependencies rclpy example_interfaces --node-name service_server_02

service_server_02.py 代码

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
# 导入接口
from example_interfaces.srv import AddTwoInts

class ServiceServer02(Node):
def __init__(self,name):
super().__init__(name)
self.get_logger().info("节点已启动：%s!" % name)
self.add_ints_server_ = self.create_service(AddTwoInts,"add_two_ints_srv", self.handle_add_two_ints)

def handle_add_two_ints(self,request, response):
self.get_logger().info(f"收到请求，计算{request.a} + {request.b}")
response.sum = request.a + request.b
return response

def main(args=None):
rclpy.init(args=args) # 初始化rclpy
node = ServiceServer02("service_server_02") # 新建一个节点
rclpy.spin(node) # 保持节点运行，检测是否收到退出指令（Ctrl+C）
rclpy.shutdown() # 关闭rclpy

2.创建服务消费者service_client_02.py

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class ServiceClient02(Node):
def __init__(self,name):
super().__init__(name)
self.get_logger().info("节点已启动：%s!" % name)
self.client_ = self.create_client(AddTwoInts,"add_two_ints_srv")

def result_callback_(self, result_future):
response = result_future.result()
self.get_logger().info(f"收到返回结果：{response.sum}")

def send_request(self, a, b):
while rclpy.ok() and self.client_.wait_for_service(1)==False:
self.get_logger().info(f"等待服务端上线....")

request = AddTwoInts.Request()
request.a = a
request.b = b
self.client_.call_async(request).add_done_callback(self.result_callback_)

def main(args=None):
rclpy.init(args=args) # 初始化rclpy
node = ServiceClient02("service_client_02") # 新建一个节点
# 调用函数发送请求
node.send_request(3,4)
rclpy.spin(node) # 保持节点运行，检测是否收到退出指令（Ctrl+C）
rclpy.shutdown() # 关闭rclpy