从 .proto 到 Python：使用 Protocol Buffers 的完整实践指南

✨ 为什么使用 Protobuf？

Protocol Buffers（简称 protobuf）是 Google 推出的高效、跨语言、结构化数据序列化协议。相比 JSON / XML，它具备：

• 📦 结构清晰：强类型 + 字段编号
• 🚀 更高效：体积小，速度快
• 🔁 跨语言：支持 Python / Java / Go / C++ 等
• 🔐 序列化 & 反序列化自动完成

---

🧱 使用步骤概览

使用 protobuf 的标准流程如下：

1. 定义数据结构（.proto 文件）
2. 使用 protoc 编译生成 Python 文件
3. 在业务代码中导入并使用

---

✍️ Step 1：定义 .proto 文件

创建 history.proto 文件，内容如下：

syntax = "proto3";
package history;message Video {string video_id = 1;float video_time = 2;float watch_time = 3;int32 event_time = 4;
}message VideoHistory {repeated Video videos = 1;
}

📌 说明：

• video_id：视频 ID
• video_time：视频总时长
• watch_time：观看时长
• event_time：发生时间（秒级时间戳）
• repeated：表示数组/列表

---

🛠️ Step 2：使用 protoc 编译生成 Python 文件

确保已安装 protoc，然后执行：

protoc --proto_path=. --python_out=./dto history.proto

执行后，自动生成 dto/history_pb2.py 文件，不要手动编辑这个文件，它是自动生成的。

---

🧪 Step 3：在 Python 中使用

from dto.history_pb2 import Video, VideoHistory# 构造单个视频数据
video = Video(video_id="vid_001",video_time=120.0,watch_time=90.0,event_time=1720001234
)# 构造历史记录
history = VideoHistory()
history.videos.append(video)# 序列化为二进制（可用于网络传输、存储）
binary_data = history.SerializeToString()# 反序列化回来
new_history = VideoHistory()
new_history.ParseFromString(binary_data)print("恢复的视频ID：", new_history.videos[0].video_id)

📦 Protobuf vs JSON 在 Redis 中的使用对比

对比项	Protobuf（二进制）	JSON（字符串）
体积	✅ 小	❌ 大
性能	✅ 快（原生序列化）	❌ 慢（字符串解析）
可读性	❌ 差	✅ 强
强类型结构	✅ 有	❌ 无
版本兼容性	✅ 好	❌ 差（需靠约定）
调试难度	❌ 高	✅ 低

🧱 Redis 存储方式：string vs list

✅ Redis key: string（整体存 protobuf）

redis.set("user_click:123", history.SerializeToString())

适合：

• 存储完整行为轨迹
• 更新频率不高，或可加锁处理

---

✅ Redis key: list（每条行为一个 protobuf）

redis.lpush("user_click:123", video.SerializeToString())
redis.ltrim("user_click:123", 0, 49)  # 限长

适合：

• 高频点击写入
• 实时滑窗建模（保留最近 N 条）
• 更易扩展为 Kafka / 日志流

---

🔁 Protobuf + Redis 封装建议

def append_video(redis, uid: str, video: Video, max_len=50):key = f"user_click:{uid}"redis.lpush(key, video.SerializeToString())redis.ltrim(key, 0, max_len - 1)def get_user_clicks(redis, uid: str, limit=50):key = f"user_click:{uid}"raw_list = redis.lrange(key, 0, limit - 1)return [deserialize_video(raw) for raw in raw_list]def deserialize_video(raw: bytes) -> Video:v = Video()v.ParseFromString(raw)return v