树莓派4（Bullseye）摄像头3 MJPG视频跨设备传输故障排查及方案咨询

我正在尝试从搭载Camera Module 3的Raspberry Pi 4（Bullseye系统）向另一台设备传输视频流，最终要用于Azure AI Vision或其他实时计算机视觉分析，因此不想仅依赖VLC这类工具。目前两台设备能建立连接且有数据传输（调试数据显示是编码后的JPEG格式：b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xdb\x00C\x00\r\t\n ...'），但流式传输无法正常工作，怀疑是逻辑或编解码环节出错，寻求排查建议。

我猜测核心问题是：客户端代码一直在寻找--FRAME分隔符和Content-Type头部，但服务端并未添加这类标记，导致接收逻辑卡死。

树莓派端生产者/编码器代码 #

import socket
import time

from picamera2 import Picamera2
from picamera2.encoders import JpegEncoder
from picamera2.outputs import FileOutput

picam2 = Picamera2()
video_config = picam2.create_video_configuration({"size": (1280, 720)})
picam2.configure(video_config)

encoder = JpegEncoder()

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("my-pi-ip" , 10001))
    sock.listen()

    picam2.encoders = encoder

    print("\nAwaiting connection...")
    conn, addr = sock.accept()
    stream = conn.makefile("wb")
    encoder.output = FileOutput(stream)
    
    data = conn.recv(1024)
    if data.decode() == "ping":
        print("ping recieved from: " + str(conn))

    picam2.start_encoder()
    picam2.start()
    print("camera started")
    
    time.sleep(15)
    
    picam2.stop()
    print("\ncamera stopped")
    picam2.stop_encoder()
    conn.close()

本地机器端消费者/解码器代码 #

import cv2
import socket
import numpy as np
import threading
import time

# Create a socket client
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(("my-pi-ip", 10001))

ping_exception_flag = False

# Function to periodically send ping to the server
def send_ping():
    print("Ping thread started")
    while True:
        try:
            print("Sending ping")
            client_socket.send(b'ping')
            print("Ping sent")
            time.sleep(5)
        except:
            print("Server disconnected")
            ping_exception_flag = True


# Start the ping thread
ping_thread = threading.Thread(target=send_ping)
ping_thread.start()

# Receive and display the video stream
while not ping_exception_flag:
    data = b''
    while b'--FRAME' not in data:
        data += client_socket.recv(1024)
        print(data)
        #debug purposes

    frames = data.split(b'--FRAME')
    print(data)

    for frame in frames:
        print(frame)
        if b'Content-Type: image/jpeg' in frame:
            image_data = frame.split(b'\r\n\r\n')[1]
            image_array = np.frombuffer(image_data, dtype=np.uint8)
            frame = cv2.imdecode(image_array, flags=cv2.IMREAD_COLOR)
            if frame is not None:
                cv2.imshow("Video Stream", frame)
                if cv2.waitKey(1) & 0xFF == ord('q'):
                    break

# Cleanup
ping_thread.join()  # Wait for the ping thread to finish
cv2.destroyAllWindows()
client_socket.close()

1. 修复帧分割逻辑（核心问题） #

服务端仅输出原始JPEG数据流，无任何帧边界标记，客户端却在等待不存在的--FRAME和Content-Type，导致接收逻辑卡死。以下两种方案可解决该问题：

方案A：添加帧长度前缀 #

服务端在每个JPEG帧前发送4字节的帧长度（大端格式），客户端先读取长度，再读取对应字节数的JPEG数据：

修改后服务端关键代码：

# ... 原有代码 ...
picam2.start()
print("camera started")

# 自定义输出回调，添加长度前缀
def send_frame(data):
    # 发送4字节帧长度
    frame_len = len(data)
    conn.sendall(frame_len.to_bytes(4, byteorder='big'))
    # 发送JPEG数据
    conn.sendall(data)

encoder.output = FileOutput(send_frame)

time.sleep(15)
# ... 原有代码 ...

修改后客户端接收逻辑：

# Receive and display the video stream
while not ping_exception_flag.is_set():
    try:
        # 读取4字节帧长度
        len_data = client_socket.recv(4)
        if not len_data:
            break
        frame_len = int.from_bytes(len_data, byteorder='big')
        
        # 读取完整JPEG帧
        image_data = b''
        while len(image_data) < frame_len:
            chunk = client_socket.recv(min(frame_len - len(image_data), 4096))
            if not chunk:
                break
            image_data += chunk
        
        # 解码显示
        image_array = np.frombuffer(image_data, dtype=np.uint8)
        frame = cv2.imdecode(image_array, flags=cv2.IMREAD_COLOR)
        if frame is not None:
            cv2.imshow("Video Stream", frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                ping_exception_flag.set()
                break
    except Exception as e:
        print(f"接收帧出错: {e}")
        ping_exception_flag.set()
        break

方案B：利用JPEG头尾标记分割 #

JPEG帧固定以\xff\xd8开头、\xff\xd9结尾，客户端可通过缓存数据并匹配这些标记来分割帧：

# 客户端接收逻辑替换为
buffer = b''
while not ping_exception_flag.is_set():
    buffer += client_socket.recv(4096)
    start_idx = 0
    while True:
        # 找JPEG起始标记
        start = buffer.find(b'\xff\xd8', start_idx)
        if start == -1:
            break
        # 找JPEG结束标记
        end = buffer.find(b'\xff\xd9', start)
        if end == -1:
            break
        # 提取完整帧并处理
        jpeg_data = buffer[start:end+2]
        image_array = np.frombuffer(jpeg_data, dtype=np.uint8)
        frame = cv2.imdecode(image_array, flags=cv2.IMREAD_COLOR)
        if frame is not None:
            cv2.imshow("Video Stream", frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                ping_exception_flag.set()
                break
        # 清理缓存，保留未处理的剩余数据
        buffer = buffer[end+2:]
        start_idx = 0

2. 修复Ping线程的变量作用域问题 #

原客户端中ping_exception_flag是局部变量，线程内修改无法同步到主线程，需改用threading.Event：

# 替换原flag定义
ping_exception_flag = threading.Event()

# 修改send_ping函数
def send_ping():
    print("Ping thread started")
    while not ping_exception_flag.is_set():
        try:
            print("Sending ping")
            client_socket.send(b'ping')
            print("Ping sent")
            time.sleep(5)
        except:
            print("Server disconnected")
            ping_exception_flag.set()

# 主循环判断条件改为
while not ping_exception_flag.is_set():

3. 修正服务端编码器设置 #

原服务端中picam2.encoders = encoder的写法不符合Picamera2规范，应改为：

# 去掉 picam2.encoders = encoder
# 启动编码器时直接传入实例
picam2.start_encoder(encoder)

如果最终要对接Azure AI Vision或其他CV框架，以下方案更高效稳定：

• RTSP流传输：用树莓派自带的libcamera-vid推RTSP流（命令示例：libcamera-vid -t 0 --inline -o - | cvlc -vvv stream:///dev/stdin --sout '#rtp{sdp=rtsp://:8554/stream}' :demux=h264），客户端直接用cv2.VideoCapture("rtsp://pi-ip:8554/stream")读取，无需手动处理socket。
• MQTT消息队列：适合低带宽或分布式场景，用Eclipse Mosquitto搭建MQTT服务，树莓派推送JPEG帧，客户端订阅后处理，方便后续对接云服务。
• 本地直连云服务：树莓派采集帧后直接调用Azure AI Vision SDK上传分析，省去中间设备的传输环节，降低延迟。

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