如何基于ROS/OpenCV/Flask实现ROS摄像头话题web视频流推流？

核心问题与优化方案 #

你遇到的随机报错基本都是原代码多线程安全、状态校验、资源调度的逻辑缺陷导致，按以下方向优化即可：

常见报错根因 #

• 初始状态下全局图像变量cv_image为整型0，第一帧未收到时cv2.imencode直接传入非图像类型数据，触发类型错误
• ROS订阅回调线程和Flask推流线程并发读写全局图像变量没有锁保护，出现数据竞态，读取到半写入的无效图像数据
• 发布端每2秒才发一帧，推流逻辑死循环重复编码旧帧，大量占用CPU，同时cv2.imencode失败后无判断直接调用tobytes触发空指针错误
• Flask开启Debug模式会启动重载进程，导致ROS节点重复初始化，出现节点冲突、订阅失效问题

优化后完整代码 #

#!/usr/bin/env python
import rospy
import threading
import cv2
import numpy as np
from sensor_msgs.msg import Image
from cv_bridge import CvBridge, CvBridgeError
from flask import Flask, Response

app = Flask(__name__)
bridge = CvBridge()

# 全局变量 + 锁 + 状态标记
cv_image = None
frame_lock = threading.Lock()
new_frame_event = threading.Event()

def callback(data):
    global cv_image
    try:
        img = bridge.imgmsg_to_cv2(data, "bgr8")
        # 加锁写图像
        with frame_lock:
            cv_image = img.copy()
        # 通知有新帧
        new_frame_event.set()
    except CvBridgeError as e:
        rospy.logerr(f"图像转换失败: {e}")

# ROS节点初始化放在主线程逻辑前，避免debug重载重复初始化
if not rospy.core.is_initialized():
    rospy.init_node('webcam_stream_node', disable_signals=True)
image_sub = rospy.Subscriber("webcam/image_raw", Image, callback)

# 生成默认占位图，未收到帧时展示
def get_placeholder():
    placeholder = np.zeros((480, 640, 3), dtype=np.uint8)
    cv2.putText(placeholder, "Waiting for camera frame...", (50, 240), 
                cv2.FONT_HERSHEY_SIMPLEX, 1, (255,255,255), 2)
    ret, jpeg = cv2.imencode('.jpg', placeholder)
    return jpeg.tobytes()

def gen():
    global cv_image
    while True:
        # 等待新帧通知，超时1秒避免卡死
        if new_frame_event.wait(timeout=1):
            new_frame_event.clear()
            # 加锁读图像
            with frame_lock:
                current_frame = cv_image.copy()
            # 编码校验
            ret, jpeg = cv2.imencode('.jpg', current_frame)
            if ret:
                frame = jpeg.tobytes()
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')
        else:
            # 超时返回占位图
            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + get_placeholder() + b'\r\n\r\n')

@app.route('/')
def video_feed():
    return Response(gen(),
                    mimetype='multipart/x-mixed-replace; boundary=frame')

if __name__ == '__main__':
    # 生产环境关闭debug，调试阶段要开debug的话加use_reloader=False避免重复启动ROS节点
    app.run(host='0.0.0.0', port=5000, debug=False)

额外优化说明 #

• 加入了线程锁保证图像读写的原子性，避免数据竞态
• 用事件通知机制替代死循环，只有收到新帧才编码，大幅降低CPU占用
• 新增占位图逻辑，未收到摄像头帧时不会直接报错
• 补全了所有可能失败的操作的异常/返回值校验，避免随机崩溃
• 优化了ROS节点初始化逻辑，避免重复初始化冲突

内容的提问来源于stack exchange，提问作者Aziz zeGeek