从“你听我说”到“我问你答”：用Python和C++手把手实现ROS2话题与服务通信
从“你听我说”到“我问你答”：用Python和C++手把手实现ROS2话题与服务通信
在机器人开发领域，通信机制如同神经系统般重要。当你已经成功运行了ROS2的talker/listener示例，却对如何在实际项目中灵活运用话题（Topic）和服务（Service）感到困惑时，这篇文章将为你拨开迷雾。我们将通过一个机器人控制场景，对比分析两种通信模式的特点，并分别用Python和C++实现完整案例。
1. 通信模型的核心差异：广播与应答
理解ROS2通信，首先要掌握话题和服务的本质区别。想象一个机器人系统：话题如同广播电台，持续发送数据给所有感兴趣的接收者；服务则像客服热线，需要明确的一问一答。
关键对比维度：
| 特性 | 话题（Topic） | 服务（Service） |
|---|---|---|
| 通信模式 | 发布/订阅（一对多） | 请求/响应（一对一） |
| 数据流向 | 单向 | 双向 |
| 实时性 | 适合持续数据流 | 适合按需请求 |
| 典型应用场景 | 传感器数据、控制指令 | 状态查询、计算服务 |
| 节点耦合度 | 低（发布者无需知道订阅者） | 高（客户端需知道服务端） |
提示：在机器人系统中，激光雷达数据通常采用话题传输，而电池状态查询则更适合服务模式。
2. Python实现：机器人控制指令发布
让我们用Python构建一个完整的机器人控制案例。假设我们需要实现：
- 话题：持续发布移动指令（前进/停止）
- 服务：响应传感器数据查询请求
2.1 话题发布者实现
创建robot_controller.py文件：
import rclpy from rclpy.node import Node from std_msgs.msg import String class RobotController(Node): def __init__(self): super().__init__('robot_controller') self.publisher = self.create_publisher(String, 'control_command', 10) self.timer = self.create_timer(1.0, self.timer_callback) self.current_command = "forward" def timer_callback(self): msg = String() msg.data = self.current_command self.publisher.publish(msg) self.get_logger().info(f'发布指令: {msg.data}') # 模拟指令切换 self.current_command = "stop" if self.current_command == "forward" else "forward" def main(args=None): rclpy.init(args=args) node = RobotController() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main()2.2 服务端实现
扩展上述代码，添加传感器查询服务：
from example_interfaces.srv import AddTwoInts class RobotController(Node): def __init__(self): # ...原有初始化代码... self.srv = self.create_service( AddTwoInts, 'query_sensor', self.handle_sensor_query) def handle_sensor_query(self, request, response): # 模拟传感器数据计算 response.sum = request.a + request.b self.get_logger().info(f'处理查询: {request.a} + {request.b}') return response3. C++实现：高效通信方案
对于性能敏感的场景，C++是更好的选择。下面展示等效的C++实现。
3.1 话题通信实现
创建robot_controller.cpp文件：
#include "rclcpp/rclcpp.hpp" #include "std_msgs/msg/string.hpp" using namespace std::chrono_literals; class RobotController : public rclcpp::Node { public: RobotController() : Node("robot_controller"), current_command_("forward") { publisher_ = this->create_publisher<std_msgs::msg::String>("control_command", 10); timer_ = this->create_wall_timer( 1000ms, std::bind(&RobotController::timer_callback, this)); } private: void timer_callback() { auto message = std_msgs::msg::String(); message.data = current_command_; publisher_->publish(message); RCLCPP_INFO(this->get_logger(), "发布指令: %s", message.data.c_str()); current_command_ = (current_command_ == "forward") ? "stop" : "forward"; } rclcpp::TimerBase::SharedPtr timer_; rclcpp::Publisher<std_msgs::msg::String>::SharedPtr publisher_; std::string current_command_; }; int main(int argc, char * argv[]) { rclcpp::init(argc, argv); rclcpp::spin(std::make_shared<RobotController>()); rclcpp::shutdown(); return 0; }3.2 服务通信实现
添加服务处理逻辑：
#include "example_interfaces/srv/add_two_ints.hpp" class RobotController : public rclcpp::Node { public: RobotController() : Node("robot_controller") { // ...原有构造函数代码... service_ = this->create_service<example_interfaces::srv::AddTwoInts>( "query_sensor", std::bind(&RobotController::handle_sensor_query, this, std::placeholders::_1, std::placeholders::_2)); } private: void handle_sensor_query( const std::shared_ptr<example_interfaces::srv::AddTwoInts::Request> request, std::shared_ptr<example_interfaces::srv::AddTwoInts::Response> response) { response->sum = request->a + request->b; RCLCPP_INFO(this->get_logger(), "处理查询: %ld + %ld", request->a, request->b); } rclcpp::Service<example_interfaces::srv::AddTwoInts>::SharedPtr service_; };4. 实战：构建完整通信系统
现在我们将Python和C++节点组合成一个完整系统，展示跨语言通信能力。
4.1 系统架构设计
- Python节点：作为控制中心，发布移动指令并响应查询
- C++节点：作为执行单元，订阅指令并调用服务
通信流程：
[Python Controller] --(control_command/topic)--> [C++ Executor] [Python Controller] <--(query_sensor/service)--- [C++ Executor]4.2 C++执行节点实现
创建robot_executor.cpp：
#include "rclcpp/rclcpp.hpp" #include "std_msgs/msg/string.hpp" #include "example_interfaces/srv/add_two_ints.hpp" class RobotExecutor : public rclcpp::Node { public: RobotExecutor() : Node("robot_executor") { // 订阅控制指令 subscription_ = this->create_subscription<std_msgs::msg::String>( "control_command", 10, std::bind(&RobotExecutor::command_callback, this, std::placeholders::_1)); // 创建服务客户端 client_ = this->create_client<example_interfaces::srv::AddTwoInts>("query_sensor"); } private: void command_callback(const std_msgs::msg::String::SharedPtr msg) { RCLCPP_INFO(this->get_logger(), "执行指令: %s", msg->data.c_str()); // 收到指令后查询传感器数据 if (msg->data == "stop") { auto request = std::make_shared<example_interfaces::srv::AddTwoInts::Request>(); request->a = 10; request->b = 20; client_->async_send_request(request, [this](rclcpp::Client<example_interfaces::srv::AddTwoInts>::SharedFuture future) { auto response = future.get(); RCLCPP_INFO(this->get_logger(), "传感器数据: %ld", response->sum); }); } } rclcpp::Subscription<std_msgs::msg::String>::SharedPtr subscription_; rclcpp::Client<example_interfaces::srv::AddTwoInts>::SharedPtr client_; };4.3 系统运行测试
- 启动Python控制节点：
python3 robot_controller.py- 启动C++执行节点：
ros2 run robot_pkg robot_executor- 观察终端输出，可以看到完整的通信流程：
- 控制指令定期发布
- 执行节点收到指令后发起服务调用
- 控制节点处理服务请求并返回结果
5. 高级技巧与性能优化
在实际项目中，通信性能往往成为瓶颈。以下是提升ROS2通信效率的关键方法：
5.1 通信质量策略（QoS）配置
ROS2允许精细控制通信质量：
# Python示例 from rclpy.qos import QoSProfile, QoSReliabilityPolicy qos_profile = QoSProfile( depth=10, reliability=QoSReliabilityPolicy.RELIABLE # 或BEST_EFFORT ) self.publisher = self.create_publisher( String, 'control_command', qos_profile=qos_profile)// C++示例 #include "rclcpp/qos.hpp" auto qos = rclcpp::QoS(10).reliable(); publisher_ = this->create_publisher<std_msgs::msg::String>( "control_command", qos);QoS关键参数对比：
| 参数 | 可靠传输(RELIABLE) | 尽力传输(BEST_EFFORT) |
|---|---|---|
| 数据保证 | 不丢失 | 可能丢失 |
| 延迟 | 较高 | 较低 |
| 带宽占用 | 较高 | 较低 |
| 适用场景 | 关键指令 | 高频传感器数据 |
5.2 零拷贝通信
对于高性能场景，可以使用零拷贝技术减少数据复制：
// C++零拷贝示例 void timer_callback() { auto loaned_msg = publisher_->borrow_loaned_message(); loaned_msg.get().data = "forward"; publisher_->publish(std::move(loaned_msg)); }5.3 多线程处理
默认情况下，ROS2使用单线程执行器。对于计算密集型任务，需配置多线程：
# Python多线程示例 import rclpy.executors executor = rclpy.executors.MultiThreadedExecutor() executor.add_node(node) executor.spin()// C++多线程示例 rclcpp::executors::MultiThreadedExecutor executor; executor.add_node(node); executor.spin();在实际机器人项目中，通信机制的选择直接影响系统性能和可靠性。通过本文的对比实现，你应该能够根据具体需求选择合适通信方式，并优化其性能。