如何序列化GStreamer的Sample对象以通过Kafka发送？

尝试用Kafka发送GStreamer的帧Sample对象时，触发序列化错误：

TypeError: cannot pickle 'Sample' object

使用的GStreamer管道：

"rtspsrc location=rtsp://admin:123456@192.168.6.28/sub latency=0 ! rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! videoscale ! video/x-raw,format=BGR,width=640,height=640  ! appsink name=sink sync=false"

报错代码片段：

sample = self.appsink.try_pull_sample(Gst.SECOND)
producer.produce("testTopic", pickle.dumps(sample), "1", callback=delivery_callback)

已尝试json.dumps、bytes、ArrayBytes等方式，均未解决问题。

GStreamer的Sample是包含缓冲区、格式元数据的复杂对象，无法直接序列化。正确思路是从Sample中提取原始图像数据和格式信息，再对这些基础数据进行序列化传输，接收端根据信息还原图像。

步骤1：从Sample中提取核心数据 #

从Sample中取出缓冲区，读取原始字节数据，同时提取图像宽、高、格式等关键信息：

sample = self.appsink.try_pull_sample(Gst.SECOND)
if sample:
    buffer = sample.get_buffer()
    # 映射缓冲区内存以读取数据
    success, map_info = buffer.map(Gst.MapFlags.READ)
    if success:
        # 原始BGR图像字节数据
        img_raw_data = map_info.data
        # 从Caps中提取图像格式参数
        caps = sample.get_caps()
        struct = caps.get_structure(0)
        width = struct.get_value("width")
        height = struct.get_value("height")
        img_format = struct.get_value("format")
        # 释放缓冲区映射
        buffer.unmap(map_info)

步骤2：封装并序列化数据 #

将原始数据和格式信息封装为可序列化结构，选择适合的序列化方式：

方式1：用Pickle序列化（Python环境下高效）

import pickle

# 封装成字典
frame_package = {
    "width": width,
    "height": height,
    "format": img_format,
    "raw_data": img_raw_data
}
# 序列化后发送到Kafka
serialized_data = pickle.dumps(frame_package)
producer.produce("testTopic", serialized_data, "1", callback=delivery_callback)

方式2：用JSON+Base64（跨语言友好）

由于JSON不支持直接存储字节数据，需先将原始数据转为Base64字符串：

import json
import base64

img_base64 = base64.b64encode(img_raw_data).decode('utf-8')
frame_package = {
    "width": width,
    "height": height,
    "format": img_format,
    "raw_data": img_base64
}
serialized_data = json.dumps(frame_package).encode('utf-8')
producer.produce("testTopic", serialized_data, "1", callback=delivery_callback)

步骤3：接收端还原图像 #

根据序列化方式反向操作，还原成可处理的图像：

Pickle方式还原

import pickle
import cv2
import numpy as np

# 解析Kafka消息
received_package = pickle.loads(message.value)
# 将字节数据转为numpy数组并reshape为图像
img_array = np.frombuffer(received_package["raw_data"], dtype=np.uint8)
img = img_array.reshape((received_package["height"], received_package["width"], 3))
# 后续处理示例
cv2.imshow("Received Frame", img)

JSON+Base64方式还原

import json
import base64
import cv2
import numpy as np

received_package = json.loads(message.value.decode('utf-8'))
img_raw_data = base64.b64decode(received_package["raw_data"])
img_array = np.frombuffer(img_raw_data, dtype=np.uint8)
img = img_array.reshape((received_package["height"], received_package["width"], 3))
cv2.imshow("Received Frame", img)

优化建议 #

如果追求传输效率，可修改GStreamer管道，直接传输H264编码后的帧（无需解码为BGR），大幅减少数据体积：

"rtspsrc location=rtsp://admin:123456@192.168.6.28/sub latency=0 ! rtph264depay ! h264parse ! appsink name=sink sync=false"

接收端再对H264数据进行解码即可，适合高帧率、大分辨率场景。

内容的提问来源于stack exchange，提问作者Suhaib Al Hindi