### ROS2 中基于 Python 的消息发布与订阅 在 ROS2 中，消息的发布与订阅是核心功能之一。以下是关于如何使用 Python 编写一个简单的发布者和订阅者的详细说明。 #### 创建 ROS2 工作空间 首先需要设置 ROS2 的工作环境并创建一个新的包： ```bash mkdir -p ~/ros2_ws/src cd ~/ros2_ws/ colcon build source install/setup.bash ``` 接着，在 `src` 文件夹下创建新的 ROS2 包： ```bash ros2 pkg create --build-type ament_python py_pubsub ``` 这将在 `py_pubsub` 路径中生成必要的文件结构。 --- #### 发布者 (Publisher) 下面是一个基本的 Python 代码示例，用于实现消息的发布： ```python import rclpy from rclpy.node import Node from std_msgs.msg import String class MinimalPublisher(Node): def __init__(self): super().__init__('minimal_publisher') self.publisher_ = self.create_publisher(String, 'topic', 10) timer_period = 0.5 # seconds self.timer = self.create_timer(timer_period, self.timer_callback) self.i = 0 def timer_callback(self): msg = String() msg.data = f'Hello World: {self.i}' self.publisher_.publish(msg) self.get_logger().info(f'Publishing: "{msg.data}"') self.i += 1 def main(args=None): rclpy.init(args=args) minimal_publisher = MinimalPublisher() rclpy.spin(minimal_publisher) minimal_publisher.destroy_node() rclpy.shutdown() if __name__ == '__main__': main() ``` 上述代码定义了一个名为 `MinimalPublisher` 的节点类，它会每隔 0.5 秒向主题 `/topic` 发送一条字符串消息[^1]。 保存此脚本到 `py_pubsub/py_pubsub/publisher_member_function.py` 并确保其可执行权限。 --- #### 订阅者 (Subscriber) 下面是对应的订阅者代码，该代码监听来自同一主题的消息，并打印接收到的内容： ```python import rclpy from rclpy.node import Node from std_msgs.msg import String class MinimalSubscriber(Node): def __init__(self): super().__init__('minimal_subscriber') self.subscription = self.create_subscription( String, 'topic', self.listener_callback, 10) self.subscription # prevent unused variable warning def listener_callback(self, msg): self.get_logger().info(f'I heard: "{msg.data}"') def main(args=None): rclpy.init(args=args) minimal_subscriber = MinimalSubscriber() rclpy.spin(minimal_subscriber) minimal_subscriber.destroy_node() rclpy.shutdown() if __name__ == '__main__': main() ``` 这段代码实现了订阅者逻辑，当检测到来自 `/topic` 主题的新消息时，调用回调函数 `listener_callback` 来处理这些数据。 将此脚本保存至 `py_pubsub/py_pubsub/subscriber_member_function.py`。 --- #### 运行测试 启动两个终端窗口分别运行以下命令以验证通信效果： **运行发布者** ```bash ros2 run py_pubsub publisher_member_function ``` **运行订阅者** ```bash ros2 run py_pubsub subscriber_member_function ``` 如果一切正常，您应该会在订阅端看到类似如下输出： ``` I heard: "Hello World: 0" I heard: "Hello World: 1" ... ``` --- #### 处理 CompressedImage 类型的消息 对于更复杂的数据类型（如图像），可以参考以下方法来订阅和发布 `sensor_msgs/msg/CompressedImage` 数据[^3]： ##### 修改 Publisher 和 Subscriber 部分代码 修改后的发布者部分可能类似于这样： ```python from sensor_msgs.msg import CompressedImage ... class ImagePublisher(Node): def __init__(self): super().__init__('image_publisher') self.publisher_ = self.create_publisher(CompressedImage, 'camera_image/compressed', 10) ... def publish_image(self, image_data): compressed_msg = CompressedImage() compressed_msg.format = "jpeg" compressed_msg.data = image_data.tobytes() # 假设输入为 NumPy 数组形式 self.publisher_.publish(compressed_msg) ``` 而订阅者则需解析压缩图像流： ```python from cv_bridge import CvBridge ... class ImageSubscriber(Node): def __init__(self): super().__init__('image_subscriber') self.bridge = CvBridge() self.subscription = self.create_subscription( CompressedImage, 'camera_image/compressed', self.image_callback, 10) def image_callback(self, msg): try: frame = self.bridge.compressed_imgmsg_to_cv2(msg, desired_encoding='bgr8') # 对帧进行进一步操作 except Exception as e: self.get_logger().error(str(e)) ``` 以上展示了如何扩展基础框架支持特定传感器数据传输。 ---