Kafka传输视频流在Python简易HTTP服务中仅显示首帧问题

代码不存在配置遗漏，核心是HTTP响应逻辑不符合视频流的传输要求：

• 现有逻辑第一次循环就发送了单张图片的HTTP响应头，后续循环重复发送响应头属于无效内容，浏览器收到image/jpeg类型的响应后，拿到第一帧数据就会判定资源加载完成，不会继续读取连接上的后续内容。
• 单张静态图片的HTTP响应天然不支持连续推送多帧，浏览器原生支持的HTTP实时视频流方案是MJPEG（Motion JPEG）格式，需要按对应规范写响应内容才能实现连续播放。

另外现有代码还有几个隐性bug：

• Producer代码缺失time模块导入，视频读取到末尾时帧为空会触发编码报错，使用完Producer未主动释放连接
• Consumer侧冗余导入了ensurepip的bootstrap方法，无实际作用
• 全局初始化KafkaConsumer会导致消费位点被第一个请求独占，后续新请求无法正常获取帧数据

1. 单独启动后台线程消费Kafka消息，用带锁的全局变量缓存最新收到的帧，避免请求处理逻辑直接消费Kafka导致多客户端访问时帧错乱
2. 接口响应头设置为multipart/x-mixed-replace类型，告知浏览器当前响应是会持续替换内容的流资源
3. 每帧数据按照MJPEG规范，用固定边界符分隔，单独标注帧类型和长度后写入响应流
4. 增加缓存禁用头，避免浏览器或代理缓存帧导致画面不更新，Kafka消费者配置从最新位点开始消费，避免服务重启后推送历史旧帧

Kafka 生产者代码 #

import cv2
import io
import time
from kafka import KafkaProducer

def recordedVid(video_file):
    producer = KafkaProducer(
        bootstrap_servers='localhost:9092'
    )
    cap = cv2.VideoCapture(video_file)
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        ret, buffer = cv2.imencode('.jpg', frame)
        toStream = buffer.tobytes()
        producer.send('Video', toStream)
        print('sent')
        time.sleep(0.1)
    cap.release()
    producer.close()
    print('done')

recordedVid('sample.mp4')

Kafka 消费者+HTTP服务代码 #

import cv2
import socket
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from kafka import KafkaConsumer
from PIL import Image
import numpy as np
import io

latest_frame = None
frame_lock = threading.Lock()

def extract_IP():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.connect(("8.8.8.8", 80))
    HOST = s.getsockname()[0]
    s.close()
    return HOST

def msg_process(msg):
    frame_bytes = msg.value
    frame_arr = np.frombuffer(frame_bytes, np.uint8)
    frame = cv2.imdecode(frame_arr, cv2.IMREAD_COLOR)
    img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    toLoad = io.BytesIO()
    img.save(toLoad, format='JPEG')
    return toLoad.getvalue()

def kafka_consumer_thread():
    global latest_frame
    consumer = KafkaConsumer(
        "Video",
        bootstrap_servers='localhost:9092',
        auto_offset_reset='latest'
    )
    for msg in consumer:
        frame_data = msg_process(msg)
        with frame_lock:
            latest_frame = frame_data

class Display(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/':
            self.send_response(200)
            self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=--frame")
            self.send_header("Cache-Control", "no-cache, private")
            self.send_header("Pragma", "no-cache")
            self.end_headers()
            while True:
                with frame_lock:
                    if latest_frame is None:
                        continue
                    frame = latest_frame
                self.wfile.write(b"--frame\r\n")
                self.send_header("Content-Type", "image/jpeg")
                self.send_header("Content-Length", str(len(frame)))
                self.end_headers()
                self.wfile.write(frame)
                self.wfile.write(b"\r\n")
        return

if __name__ == "__main__":
    consumer_thread = threading.Thread(target=kafka_consumer_thread, daemon=True)
    consumer_thread.start()
    HOST = extract_IP()
    PORT = 9999
    server = HTTPServer((HOST, PORT), Display)
    print("server is now running in ", HOST, " and Port ", PORT)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    server.server_close()

启动生产者和服务后，直接在浏览器访问http://<服务IP>:9999/即可看到持续更新的视频流，该实现支持多客户端同时访问，所有客户端都会同步拉取到最新帧。

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