Python 中使用 Protocol Buffers 完整指南

概述

Protocol Buffers (简称 Protobuf) 是 Google 开发的一种语言中立、平台中立、可扩展的序列化结构化数据的方式。本指南将带你了解如何在 Python 中使用 Protobuf。

为什么选择 Protobuf？

高效：比 JSON 更小更快
跨语言：支持多种编程语言
类型安全：编译时类型检查
向后兼容：支持 schema 演进

环境准备

安装依赖

pip install protobuf protoc-compiler

验证安装

python -c "import google.protobuf; print('Protobuf 安装成功')"

定义 Protocol Buffers

创建 .proto 文件

创建一个名为 person.proto 的文件：

syntax = "proto3";



package tutorial;



message Person {

  string name = 1;

  int32 id = 2;

  string email = 3;

  

  enum PhoneType {

    MOBILE = 0;

    HOME = 1;

    WORK = 2;

  }

  

  message PhoneNumber {

    string number = 1;

    PhoneType type = 2;

  }

  

  repeated PhoneNumber phones = 4;

}



message AddressBook {

  repeated Person people = 1;

}

编译 .proto 文件

使用 protoc 编译器

protoc --python_out=. person.proto

这将会生成 person_pb2.py 文件。

使用 grpcio-tools（推荐）

pip install grpcio-tools

python -m grpc_tools.protoc --python_out=. person.proto

基础使用

创建和序列化消息

import person_pb2



# 创建 Person 实例

person = person_pb2.Person()

person.name = "张三"

person.id = 1234

person.email = "[email protected]"



# 添加电话号码

phone = person.phones.add()

phone.number = "13800138000"

phone.type = person_pb2.Person.MOBILE



# 序列化到字节

serialized_data = person.SerializeToString()

print(f"序列化后大小: {len(serialized_data)} 字节")

反序列化消息

# 从字节反序列化

parsed_person = person_pb2.Person()

parsed_person.ParseFromString(serialized_data)



print(f"姓名: {parsed_person.name}")

print(f"ID: {parsed_person.id}")

print(f"邮箱: {parsed_person.email}")

print(f"电话: {parsed_person.phones[0].number}")

高级用法

处理多个消息

# 创建地址簿

address_book = person_pb2.AddressBook()



# 添加多个人员

for i in range(3):

    person = address_book.people.add()

    person.name = f"用户{i+1}"

    person.id = 1000 + i

    person.email = f"user{i+1}@example.com"



# 序列化整个地址簿

address_book_data = address_book.SerializeToString()

文件读写

# 写入文件

with open("address_book.bin", "wb") as f:

    f.write(address_book_data)



# 从文件读取

with open("address_book.bin", "rb") as f:

    loaded_book = person_pb2.AddressBook()

    loaded_book.ParseFromString(f.read())

    

print(f"共加载 {len(loaded_book.people)} 个人员信息")

JSON 互转

# 转换为 JSON

from google.protobuf.json_format import MessageToJson



json_str = MessageToJson(person)

print(json_str)



# 从 JSON 创建

from google.protobuf.json_format import Parse



json_person = Parse(json_str, person_pb2.Person())

最佳实践

字段验证

def validate_person(person):

    """验证 Person 消息"""

    if not person.name:

        raise ValueError("姓名不能为空")

    if person.id <= 0:

        raise ValueError("ID 必须为正数")

    if "@" not in person.email:

        raise ValueError("邮箱格式不正确")

    return True



# 使用验证

try:

    validate_person(person)

    print("验证通过")

except ValueError as e:

    print(f"验证失败: {e}")

默认值处理

# Protobuf 字段默认值

empty_person = person_pb2.Person()

print(f"默认姓名: '{empty_person.name}'")  # 空字符串

print(f"默认ID: {empty_person.id}")       # 0



# 检查字段是否设置

if empty_person.HasField("name"):

    print("姓名已设置")

else:

    print("姓名未设置")

性能优化

import time



# 批量处理

persons = []

for i in range(1000):

    p = person_pb2.Person()

    p.name = f"用户{i}"

    p.id = i

    p.email = f"user{i}@example.com"

    persons.append(p)



# 测量序列化性能

start = time.time()

for p in persons:

    p.SerializeToString()

print(f"序列化1000个对象耗时: {time.time() - start:.3f}秒")

错误处理

常见错误及解决方案

try:

    # 尝试反序列化无效数据

    invalid_data = b"invalid protobuf data"

    person = person_pb2.Person()

    person.ParseFromString(invalid_data)

except Exception as e:

    print(f"反序列化错误: {e}")



# 处理缺失字段

person = person_pb2.Person()

person.name = "测试用户"

# 不设置 email 字段

print(f"邮箱(可能为空): '{person.email}'")

完整示例

地址簿管理器

import os

import person_pb2



class AddressBookManager:

    def __init__(self, filename="address_book.bin"):

        self.filename = filename

        self.address_book = self.load_address_book()

    

    def load_address_book(self):

        """加载地址簿"""

        if os.path.exists(self.filename):

            with open(self.filename, "rb") as f:

                book = person_pb2.AddressBook()

                book.ParseFromString(f.read())

                return book

        return person_pb2.AddressBook()

    

    def add_person(self, name, person_id, email, phones=None):

        """添加新人员"""

        person = self.address_book.people.add()

        person.name = name

        person.id = person_id

        person.email = email

        

        if phones:

            for number, phone_type in phones:

                phone = person.phones.add()

                phone.number = number

                phone.type = phone_type

        

        self.save()

        return person

    

    def find_person(self, name):

        """按姓名查找人员"""

        for person in self.address_book.people:

            if person.name == name:

                return person

        return None

    

    def save(self):

        """保存地址簿"""

        with open(self.filename, "wb") as f:

            f.write(self.address_book.SerializeToString())

    

    def list_people(self):

        """列出所有人员"""

        return self.address_book.people



# 使用示例

if __name__ == "__main__":

    manager = AddressBookManager()

    

    # 添加人员

    manager.add_person(

        "李四", 

        1001, 

        "[email protected]",

        [("13900139000", person_pb2.Person.MOBILE)]

    )

    

    # 查找人员

    person = manager.find_person("李四")

    if person:

        print(f"找到: {person.name} - {person.email}")

    

    # 列出所有人员

    for p in manager.list_people():

        print(f"{p.name}: {p.email}")

总结

通过本指南，你已经学会了：

如何安装和配置 Python 的 Protobuf 环境
如何定义 .proto 文件并编译成 Python 代码
如何创建、序列化和反序列化 Protobuf 消息
如何处理复杂的数据结构
最佳实践和性能优化技巧

Protobuf 在 Python 中的使用非常直观，结合类型安全和高效性能，是构建高性能应用的理想选择。