先说一下为什么要学ros2首先他的通信非常快,而且可以多语言编程。比如说如果要采集一张然后多个模型推理然后结果汇总如果就单纯的用python的多线程多进程不仅速度慢各种损耗gil限制而且多进程多了管理起来比较复杂并且ipc开销高速度也不是很快。如果使用c如果模型中还要加几个大模型各种算子没有要自己写如果采用部分模型c推理部分模型python推理然后http通信但是这样实时性又不好。采用ros2可以就可以非常好的解决这些问题cpython多语言编程且通信速度非常快。没有ros2环境需要根据系统版本安装对应的ros2.# 安装 cv_bridge sudo apt install ros-jazzy-cv-bridge # 安装 OpenCV Python 包推荐使用 pip 获取最新版本 pip install opencv-python1创建工作空间mkdir -p ~/ros2_ws/src cd ~/ros2_ws2, 创建python包包名我们取为my_camera_pkgcd src ros2 pkg create --build-type ament_python my_camera_pkg --dependencies rclpy sensor_msgs cv_bridge参数说明--build-type ament_pythonpython包。--dependencies自动在package.xml和setup.py中添加依赖。3编辑节点代码进入包目录cd my_camera_pkg/my_camera_pkg创建两个python文件1camera_publisher.py (采集节点import rclpy from rclpy.node import Node from sensor_msgs.msg import Image import cv2 from cv_bridge import CvBridge class CameraPublisher(Node): def __init__(self): super().__init__(camera_publisher) self.publisher_ self.create_publisher(Image, /camera/image_raw, 10) self.timer self.create_timer(1.0 / 30, self.timer_callback) # 30 Hz self.cap cv2.VideoCapture(0) if not self.cap.isOpened(): self.get_logger().error(Cannot open camera) raise RuntimeError(Camera open failed) self.bridge CvBridge() self.get_logger().info(Camera publisher started) def timer_callback(self): ret, frame self.cap.read() if ret: msg self.bridge.cv2_to_imgmsg(frame, encodingbgr8) self.publisher_.publish(msg) else: self.get_logger().warn(Failed to capture frame) def destroy_node(self): self.cap.release() super().destroy_node() def main(argsNone): rclpy.init(argsargs) node CameraPublisher() try: rclpy.spin(node) except KeyboardInterrupt: pass finally: node.destroy_node() rclpy.shutdown() if __name__ __main__: main()2canny_subscriber.py推理节点,为了减少搭建环境的工作量我这里不使用yolo,用canny算子来演示。import rclpy from rclpy.node import Node from sensor_msgs.msg import Image import cv2 from cv_bridge import CvBridge class CannySubscriber(Node): def __init__(self): super().__init__(canny_subscriber) self.subscription self.create_subscription( Image, /camera/image_raw, self.image_callback, 10) self.bridge CvBridge() # 创建显示窗口 cv2.namedWindow(Canny Edge, cv2.WINDOW_NORMAL) self.get_logger().info(Canny subscriber started) def image_callback(self, msg): try: cv_image self.bridge.imgmsg_to_cv2(msg, desired_encodingbgr8) except Exception as e: self.get_logger().error(fcv_bridge error: {e}) return # 转为灰度图 gray cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) # Canny 边缘检测 edges cv2.Canny(gray, 50, 150) # 显示结果 cv2.imshow(Canny Edge, edges) cv2.waitKey(1) def destroy_node(self): cv2.destroyAllWindows() super().destroy_node() def main(argsNone): rclpy.init(argsargs) node CannySubscriber() try: rclpy.spin(node) except KeyboardInterrupt: pass finally: node.destroy_node() rclpy.shutdown() if __name__ __main__: main()4 添加可执行权限chmod x camera_publisher.py canny_subscriber.py4. 配置包信息在~/ros2_ws/src/my_camera_pkg/setup.py中找到entry_points部分。修改成entry_points{ console_scripts: [ camera_publisher my_camera_pkg.camera_publisher:main, canny_subscriber my_camera_pkg.canny_subscriber:main, ], },5. 编译工作空间#回到工作空间根目录 cd ~/ros2_ws # 编译 colcon build --packages-select my_camera_pkg source install/setup.bash #运行采集节点代码 source ~/workspaces/ros2_ws/install/setup.bash ros2 run my_camera_pkg camera_publisher6运行推理节点代码source ~/workspaces/ros2_ws/install/setup.bash ros2 run my_camera_pkg canny_subscriber7结果