Protobuf从零到实战：Python、Java与gRPC八讲教程

Protobuf 如何把对象变成二进制？为什么整数 150 会编码成 96 01？同一份 .proto 又怎样让 Python 和 Java 互相读取数据？

这篇文章按八讲整理学习过程：先跑通序列化，再理解二进制编码和版本兼容，最后通过两个完整项目完成跨语言文件交换与 gRPC 调用。示例以 proto3 为主，适合有 Python 或 Java 基础、准备接触 RPC 的读者。

学习路线与项目下载

章节 重点
第一至三讲 序列化概念、proto3 语法、Python 消息读写
第四讲 enum、map、嵌套消息与 oneof
第五讲 Tag、Varint、ZigZag 与 Wire Format
第六讲 reserved、未知字段和版本兼容
第七讲 Python / Java 订单二进制交换
第八讲 Python gRPC 服务与 Python / Java 客户端

配套项目下载：

分别解压后，在对应的 protobuf_lesson7/ 或 protobuf_lesson8/ 根目录运行命令。两个项目的 Schema 和 Java 包名各自独立，不能混用。第八讲配套版修正了 Java Deadline 的设置位置，每次 RPC 都重新设置截止时间，文中代码与下载包一致。

环境准备

Python 3.10+、JDK 17+、Maven 3.8+；第八讲需要本机空闲端口 50051。首次准备依赖需要访问 Python 包仓库和 Maven 仓库。

如果使用 Conda，可以先激活已有环境：

1
2
3
4
5
conda activate LeafMing
python --version
protoc --version
java -version
mvn -v

文中的 python 指当前环境的解释器。已具备依赖时直接使用；缺少依赖时，再根据对应项目的 requirements.txt 准备。

工具 职责
protobuf Python 消息运行时
protoc 根据 Schema 生成代码
grpcio-tools Python 可调用的编译器与 gRPC 生成插件
grpcio Python gRPC 运行时

示例通常使用 python -m grpc_tools.protoc。已有独立 protoc 时，普通消息生成命令可以改为：

1
protoc -I. --python_out=. student.proto

--python_out 生成消息类；gRPC Stub 还需要相应生成插件。编译器和运行时要符合各语言的兼容要求，不同语言的版本号不要求逐字相同。本文采用 proto3；Editions 体系可在掌握基础后继续阅读。

第 01 讲：Protobuf、JSON 与序列化

1.1 数据为什么要序列化？

假设 Python 程序有一个学生对象：

1
2
3
4
5
student = {
    "id": 1001,
    "name": "小明",
    "age": 18,
}

这个字典是 Python 进程中的对象，不能要求 Java、Go 或另一台服务器直接理解其内存结构。

因此，需要把对象转换成一串可以存储或传输的数据：

1
Python 对象 → 序列化 → bytes → 文件/网络 → 反序列化 → 新对象
  • 序列化（Serialization）：把内存中的消息转换为字节序列。
  • 反序列化（Deserialization）：按照约定的结构，把字节恢复为消息对象。

JSON 是文本序列化格式，Protobuf（Protocol Buffers）则是一种带有结构定义的二进制序列化机制。

1.2 Protobuf 与 JSON 的对比

方面 JSON Protobuf
数据形态 文本 通常是二进制
人类可读性 高 原始 bytes 不容易直接阅读
Schema 可选 通常先定义 .proto
数据体积 一般较大 在很多结构化数据场景下更紧凑
编解码性能 与实现、数据形态有关 许多场景下效率较高
类型约束 依赖额外验证 内建字段类型与编号
跨语言 支持 支持
常见用途 Web API、配置 RPC、内部服务、消息存储

Protobuf 不意味着总是比 JSON 快或小；真实表现与数据内容、实现及压缩方式有关，需要基准测试验证。

1.3 第一个 .proto

创建 student.proto：

1
2
3
4
5
6
7
syntax = "proto3";

message Student {
  int32 id = 1;
  string name = 2;
  int32 age = 3;
}

这里：

  • syntax = "proto3"; 选择 proto3 语法。
  • message Student 定义一种消息类型。
  • int32 与 string 是字段类型。
  • id、name、age 是字段名称。
  • = 1、= 2、= 3 是字段编号，不是初始值。

例如 int32 age = 3; 不是说年龄默认为 3，而是说字段 age 的编号是 3。

1.4 从 .proto 到 Python

运行：

1
python -m grpc_tools.protoc -I. --python_out=. student.proto

生成 student_pb2.py。然后：

1
2
3
4
5
6
7
8
9
10
11
12
from student_pb2 import Student

student = Student(id=1001, name="小明", age=18)
raw = student.SerializeToString()

restored = Student.FromString(raw)

print(type(raw))          # <class 'bytes'>
print(restored.id)        # 1001
print(restored.name)      # 小明
print(restored.age)       # 18
print(student == restored)  # True

第一讲的核心流程：

1
2
定义 .proto → protoc 生成代码 → 创建消息 → SerializeToString()
→ bytes → FromString()/ParseFromString() → 恢复消息

第 02 讲：proto3 语法、字段类型和编号

2.1 一个更完整的 Schema

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
syntax = "proto3";

package school;

option java_package = "com.example.school";
option java_multiple_files = true;

message Student {
  int64 id = 1;
  string name = 2;
  optional int32 age = 3;
  bool is_active = 4;
  repeated string hobbies = 5;
  optional string email = 6;
}
  • package school; 是 Protobuf 命名空间，完整消息名为 school.Student。
  • java_package 指定生成的 Java 类所在包。
  • java_multiple_files = true 使 Java 生成器创建多个顶层类型。
  • Python 模块路径并不是简单由 package 决定的，它与 .proto 路径和生成命令有关。

2.2 基本字段类型

.proto 类型 Python 类型 Java 典型类型 说明
int32 int int 有符号 32 位整数
int64 int long 有符号 64 位整数
uint32 int int 无符号 32 位，Java 需要注意无符号语义
uint64 int long 无符号 64 位，Java 需要注意无符号语义
sint32 int int 使用 ZigZag 以便高效编码小负数
float float float IEEE 754 单精度
double float double IEEE 754 双精度
bool bool boolean 布尔值
string str String UTF-8 文本
bytes bytes ByteString 任意二进制内容

金额建议：对于分作为最小计价单位的业务，通常使用 int64 price_cents，而不是用 double 累积精确金额。例如 59.90 元对应 5990 分。

2.3 字段编号的规则

1
2
3
4
message Example {
  int32 id = 1;
  string name = 2;
}

规则：

  1. 在同一个 message 中，字段编号必须唯一。
  2. 编号不能为 0，合法范围为 1 至 536870911，其中 19000 至 19999 是保留区间。
  3. 编号一旦随公开或持久化的数据投入使用，就不要修改或复用。
  4. 编号 1～15 的常见字段 Tag 编码通常只需要一个字节，因此可优先分配给高频字段。

不同的消息可以分别使用字段编号 1；这个编号并非整个项目范围内全局唯一。

2.4 默认值与 optional

对于普通的 proto3 标量字段：

1
2
3
4
5
message Person {
  int32 age = 1;
  string name = 2;
  bool active = 3;
}

Python：

1
2
3
4
person = Person()
print(person.age)      # 0
print(person.name)     # ""
print(person.active)   # False

默认读取值通常为：

类型 未设置时读取值
整数 0
浮点数 0.0
布尔 False
字符串 ""
bytes b""
repeated 空容器
map 空映射

对于普通的 int32 age = 1;，Person() 和 Person(age=0) 在隐式存在性语义下通常无法区分。

使用 optional：

1
2
3
message Person {
  optional int32 age = 1;
}
1
2
3
4
5
a = Person()
b = Person(age=0)

print(a.age, a.HasField("age"))  # 0 False
print(b.age, b.HasField("age"))  # 0 True

即使显式设置的是默认值，optional 也会跟踪“已设置”状态；在 proto3 中为新标量字段考虑 optional 是有益的设计习惯。

不要对普通的 proto3 非 optional 标量字段直接调用 HasField()；这会报错。消息字段和 oneof 成员自身也支持存在性判断。

2.5 repeated

1
2
3
message Student {
  repeated string hobbies = 1;
}
1
2
3
4
5
6
student = Student()
student.hobbies.append("编程")
student.hobbies.extend(["篮球", "阅读"])

print(list(student.hobbies))
# ['编程', '篮球', '阅读']

repeated 表示零个或多个同类型元素，并保留重复字段元素的顺序。它不是完全普通的 Python list，通常通过 append()、extend()、add() 等 API 修改。

2.6 第二讲练习：图书模型

需求：图书 ID、书名、价格、是否可借、多个作者以及可检查是否填写的 ISBN。

参考答案：

1
2
3
4
5
6
7
8
9
10
11
syntax = "proto3";
package library;

message Book {
  int32 id = 1;
  string title = 2;
  double price = 3;             // 练习要求为 double
  bool is_available = 4;
  repeated string authors = 5;
  optional string isbn = 6;
}

如果是正式电商订单，应重新审视 double price 是否符合精度要求。

第 03 讲：Python 序列化与反序列化

继续使用上一讲的 Book。

3.1 生成代码

项目：

1
2
3
4
5
book_demo/
├── book.proto
├── book_pb2.py      # protoc 生成
├── main.py
└── book.bin         # 程序运行后生成

在 book.proto 所在目录执行：

1
python -m grpc_tools.protoc -I. --python_out=. book.proto

3.2 完整的 main.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from pathlib import Path
from book_pb2 import Book


def main():
    # 1. 创建消息
    book = Book(
        id=1001,
        title="Python 入门",
        price=59.9,
        is_available=True,
        isbn="9781234567897",
    )
    book.authors.extend(["张三", "李四"])

    print("原始消息：")
    print(book)

    # 2. 序列化：Book -> bytes
    raw = book.SerializeToString()
    print("二进制：", raw)
    print("十六进制：", raw.hex(" "))
    print("字节数：", len(raw))

    # 3. 保存文件
    path = Path("book.bin")
    path.write_bytes(raw)

    # 4. 读取文件
    loaded = path.read_bytes()

    # 5. 反序列化：bytes -> Book
    restored = Book.FromString(loaded)

    print("书名：", restored.title)
    print("作者：", list(restored.authors))
    print("是否填写 ISBN：", restored.HasField("isbn"))
    print("内容相等：", book == restored)
    print("是否同一个 Python 对象：", book is restored)


if __name__ == "__main__":
    main()

核心输出：

1
2
3
4
5
书名： Python 入门
作者： ['张三', '李四']
是否填写 ISBN： True
内容相等： True
是否同一个 Python 对象： False

3.3 常用 API

操作 Python API
创建消息 Book()
普通字段赋值 book.id = 1
添加字符串数组 book.authors.append("A")
批量添加 book.authors.extend(["A", "B"])
存在性判断 book.HasField("isbn")
序列化 book.SerializeToString()
解析为新消息 Book.FromString(data)
解析到已有对象 book.ParseFromString(data)
拷贝消息内容 book.CopyFrom(other)
转十六进制 data.hex(" ")

注意两个不同层次：

  • protoc：根据 .proto 生成程序代码。
  • SerializeToString()：程序运行时把某条具体消息编码成字节。

print(book) 显示的是消息的可读文本表示，而非真正的二进制数据。

3.4 保存多个消息时的边界问题

如果顺序写入：

1
2
f.write(book1.SerializeToString())
f.write(book2.SerializeToString())

接收者无法仅凭普通 Protobuf 消息结构判断第一条消息在哪里结束、第二条消息在哪里开始。通常需要自定义长度前缀、记录分帧或使用 gRPC 等传输协议。

3.5 常见错误

错误：忘记编译 .proto。

1
ModuleNotFoundError: No module named 'book_pb2'

检查是否生成 book_pb2.py、文件是否在 Python 模块搜索路径内。

错误：试图直接替换 repeated 容器。

1
book.authors = ["A", "B"]  # 不建议，生成容器通常不允许直接重新赋值

应当调用 .extend() 或 .append()。

错误：把 bytes 当作文本写入。

1
2
with open("book.bin", "w") as f:
    f.write(book.SerializeToString())  # 类型不匹配

应使用二进制模式 wb 和 rb，或者 Path.write_bytes() / Path.read_bytes()。

第 04 讲：enum、map、嵌套消息和 oneof

现实数据通常不只是简单的整数和字符串。这一讲使用图书馆管理系统解释四种重要结构。各章模型是独立练习，不同目录下同名消息不应视为可以直接互换的版本。

4.1 enum：有限状态

1
2
3
4
5
6
enum BookStatus {
  BOOK_STATUS_UNSPECIFIED = 0;
  BOOK_STATUS_AVAILABLE = 1;
  BOOK_STATUS_BORROWED = 2;
  BOOK_STATUS_MAINTENANCE = 3;
}

proto3 枚举第一个成员必须使用编号 0。给 0 一个 UNSPECIFIED 或 UNKNOWN 含义，通常比把它当成有效业务状态更稳妥。

Python：

1
2
3
4
5
from library_pb2 import BookStatus

print(BookStatus.BOOK_STATUS_AVAILABLE)      # 1
print(BookStatus.Name(1))                    # BOOK_STATUS_AVAILABLE
print(BookStatus.Value("BOOK_STATUS_BORROWED"))  # 2

注意：proto3 枚举具有开放枚举语义。现代 Python 生成代码可能允许赋值尚未在枚举中定义的合法整数，例如 99，以支持前向兼容。enum 不等于完整的业务校验器。

4.2 map：键值映射

1
2
3
message Library {
  map<string, int32> inventory = 1;
}
1
2
3
4
5
library.inventory["1001"] = 10
library.inventory["1002"] = 5

print(dict(library.inventory))
print(library.inventory.get("missing"))  # None

一个特殊陷阱：

1
print(library.inventory["missing"])  # 0

在 Python Protobuf 的标量值 map 中，索引一个不存在的键可能新增这个键，并赋给它默认值。如果只是查询，请先用 .get() 或 in 判断。

map 键允许整数、布尔和字符串等指定标量类型，不允许浮点数、bytes、枚举、消息作为键；值可以使用消息类型，但不能直接嵌套 map。

4.3 message：组合复杂对象

1
2
3
4
5
6
7
8
9
message Author {
  string name = 1;
  string country = 2;
}

message Book {
  string title = 1;
  repeated Author authors = 2;
}
1
2
3
4
5
6
7
book = Book(title="Protobuf 实战")

author = book.authors.add()
author.name = "张三"
author.country = "China"

book.authors.add(name="Alice", country="USA")

对于 repeated string，常用 .append("张三")；对于 repeated Author，可使用 .add() 创建并加入子消息。

也可把 Author 真正声明在 Book 内部：

1
2
3
4
5
6
message Book {
  message Author {
    string name = 1;
  }
  repeated Author authors = 1;
}

此时 Python 可以使用 Book.Author(...)。如果 Author 将被多种消息共享，放在顶层通常更方便。

4.4 oneof：互斥字段

1
2
3
4
5
6
7
8
message User {
  int64 id = 1;

  oneof contact {
    string email = 2;
    string phone = 3;
  }
}
1
2
3
4
5
6
7
user = User(id=1)
user.email = "tom@example.com"
print(user.WhichOneof("contact"))  # email

user.phone = "13800138000"
print(user.WhichOneof("contact"))  # phone
print(user.email)                  # ""

oneof 同组中最多只有一个字段处于已设置状态。即使把 email 设置为空字符串，也意味着选中了 email，并会清除其他成员。

oneof 与 optional 的区别：

optional oneof
主要目的 追踪单个字段是否设置 一组字段互斥
能否判断存在性 可以 可以
多个字段可同时设置 可以 同组最多一个
示例 年龄可能没有填写 邮箱与手机号二选一

4.5 完整图书馆模型

创建 library.proto：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
syntax = "proto3";
package library;

enum BookStatus {
  BOOK_STATUS_UNSPECIFIED = 0;
  BOOK_STATUS_AVAILABLE = 1;
  BOOK_STATUS_BORROWED = 2;
  BOOK_STATUS_MAINTENANCE = 3;
}

message Author {
  string name = 1;
  string country = 2;
}

message Book {
  int32 id = 1;
  string title = 2;
  int64 price_cents = 3;
  BookStatus status = 4;
  repeated Author authors = 5;
  optional string isbn = 6;
}

message User {
  int32 id = 1;
  string name = 2;

  oneof contact {
    string email = 3;
    string phone = 4;
  }
}

message Library {
  string name = 1;
  repeated Book books = 2;
  map<string, int32> inventory = 3;
  repeated User users = 4;
}

生成：

1
python -m grpc_tools.protoc -I. --python_out=. library.proto

运行 Python：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from library_pb2 import Library, BookStatus

library = Library(name="城市图书馆")

book = library.books.add()
book.id = 1001
book.title = "Python 入门"
book.price_cents = 5990
book.status = BookStatus.BOOK_STATUS_AVAILABLE
book.authors.add(name="张三", country="China")
book.isbn = "9781234567897"

library.inventory[str(book.id)] = 10

user = library.users.add()
user.id = 1
user.name = "Tom"
user.email = "tom@example.com"

data = library.SerializeToString()
restored = Library.FromString(data)

assert library == restored
print(restored.name)
print(restored.books[0].title)
print(restored.inventory.get("1001"))
print(restored.users[0].WhichOneof("contact"))

预期：

1
2
3
4
城市图书馆
Python 入门
10
email

4.6 第四讲挑战题：电商订单 Schema 标准答案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
syntax = "proto3";
package ecommerce;

enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_PAID = 2;
  ORDER_STATUS_CANCELLED = 3;
}

message Product {
  int32 id = 1;
  string name = 2;
  int64 price_cents = 3;
}

message Order {
  int64 id = 1;
  OrderStatus status = 2;
  repeated Product products = 3;
  map<string, int32> quantities = 4;

  oneof contact {
    string email = 5;
    string phone = 6;
  }

  optional string note = 7;
}

Python 参考：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from order_pb2 import Order, OrderStatus

order = Order(id=10001, status=OrderStatus.ORDER_STATUS_PENDING)
order.products.add(id=101, name="Python 教程", price_cents=2999)
order.products.add(id=102, name="Java 教程", price_cents=5999)

order.quantities["101"] = 2
order.quantities["102"] = 1

order.email = "tom@example.com"
assert order.WhichOneof("contact") == "email"

order.phone = "13800138000"
assert order.WhichOneof("contact") == "phone"
assert not order.HasField("email")
assert not order.HasField("note")

total_cents = sum(
    p.price_cents * order.quantities.get(str(p.id), 0)
    for p in order.products
)
print(total_cents)  # 11997

restored = Order.FromString(order.SerializeToString())
assert restored == order

业务建模提醒：以上 quantities 是为练习 map 而设计。更正规的订单模型往往把每条订单行设计成独立的 OrderItem，把商品 ID、成交价和购买数量放在一起，从而避免商品列表与数量映射不同步。

第 05 讲：Wire Format、Tag、Varint 与 ZigZag

前四讲会用 Protobuf，这一讲要解释它为什么会产生那些字节。

5.1 bit、byte、十六进制

  • 1 bit：一个二进制位，值为 0 或 1。
  • 1 byte：8 bit。
  • 2 个十六进制数字恰好表示 1 byte。

例如：

1
2
3
十进制 8   = 二进制 00001000 = 十六进制 08
十进制 18  = 二进制 00010010 = 十六进制 12
十进制 150 = 二进制 10010110 = 十六进制 96

08 96 01 是三个字节，不是三个十进制数。

5.2 Wire Format 的基本结构

Protobuf 二进制消息主要由多条字段记录组成，每条字段通常是：

1
Tag + 按 Wire Type 编码的 Value

Wire Type 表：

Wire Type 名称 常见字段
0 VARINT int32、int64、uint32、bool、enum、sint32
1 I64 fixed64、sfixed64、double
2 LEN string、bytes、子消息、packed 数组
3 SGROUP 旧式 Group 开始
4 EGROUP 旧式 Group 结束
5 I32 fixed32、sfixed32、float

3、4 是旧式 Group 编码，普通新项目无需主动使用。

注意：Wire Type 不是 .proto 的字段类型。 例如 int32、bool、enum 都可能使用 Wire Type 0；仅凭 0 无法知道具体字段语义，解析还需要 Schema。

5.3 Tag 公式与推导

字段 Tag（也称 key）的数值：

1
tag = (field_number << 3) | wire_type

其中 << 3 表示左移 3 位，等价于乘以 8；| 表示按位或。对合法 Wire Type：

1
tag = field_number * 8 + wire_type

例一：

1
int32 count = 1;

field_number = 1，wire_type = 0：

1
tag = 1 * 8 + 0 = 8 = 0x08

例二：

1
string name = 2;

field_number = 2，wire_type = 2：

1
tag = 2 * 8 + 2 = 18 = 0x12

反过来拿到 Tag 的数值时：

1
2
field_number = tag >> 3
wire_type = tag & 0x07

注意：Tag 本身也采用 Varint 编码，并不是所有 Tag 都只有一个字节。

5.4 Varint：每 7 位分成一组

Varint 是可变长度整数编码。每个字节：

1
2
最高位 1 bit：后面是否还有同一整数的字节
最低 7 bit：本组有效数值
  • 最高位 1：后面还有数据。
  • 最高位 0：当前字节是最后一个。

对于非负整数：

数值 Varint 十六进制 字节数
5 05 1
127 7F 1
128 80 01 2
150 96 01 2
255 FF 01 2
300 AC 02 2
16384 80 80 01 3

手算 150：

1
2
3
4
5
6
7
8
9
150 = 22 + 1 * 128

低 7 位 = 22 = 二进制 0010110
后面还有一组，置续接位为 1 → 10010110 = 0x96

高 7 位 = 1 = 二进制 0000001
这是最后一组，续接位为 0 → 00000001 = 0x01

Varint(150) = 96 01

如果使用：

1
2
3
message Counter {
  int32 count = 1;
}

则字段 count=150 的完整编码为：

1
2
3
4
08 96 01
│  └───┘
│   value=150 (Varint)
└────── Tag: 字段 1，Wire Type 0

这就是 Protobuf 教材中最经典的三个字节。

5.5 Python 手写 Varint 编解码

为了理解机制，可以不依赖 Protobuf 运行时，写一个教学版编码器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def encode_varint(value: int) -> bytes:
    if value < 0:
        raise ValueError("这里只编码非负整数")

    result = bytearray()
    while value >= 128:
        result.append((value & 0x7F) | 0x80)
        value >>= 7
    result.append(value)
    return bytes(result)


def decode_varint(data: bytes, offset: int = 0):
    """返回 (解码整数, 下一未读取字节偏移)。
    教学实现，不代替生产级解析器的全部合法性检查。
    """
    value = 0
    shift = 0

    for pos in range(offset, len(data)):
        b = data[pos]
        value |= (b & 0x7F) << shift

        if not (b & 0x80):
            return value, pos + 1

        shift += 7
        if shift >= 70:
            raise ValueError("Varint 过长")

    raise ValueError("数据截断")


for n in (5, 127, 128, 150, 300, 16384):
    raw = encode_varint(n)
    restored, end = decode_varint(raw)
    print(n, raw.hex(" "), restored, end)

关键操作：

1
2
3
value & 0x7F     # 提取最低 7 位
| 0x80           # 置续接标记
value >>= 7      # 去掉已编码的 7 位

5.6 int32 与 sint32：ZigZag 为什么存在？

普通 int32 / int64 负数采用的 Varint 表示会进行符号扩展。以 -1 为例，数值部分占 10 字节：

1
FF FF FF FF FF FF FF FF FF 01

加上字段 1 的 Tag：

1
08 FF FF FF FF FF FF FF FF FF 01

整条字段记录占 11 字节。

如果业务数据可能经常出现绝对值很小的负数，可考虑：

1
sint32 temperature = 1;

sint32 在 Varint 之前先进行 ZigZag 映射：

原值 ZigZag 结果
0 0
-1 1
1 2
-2 3
2 4
-3 5
3 6

数学表达式：

1
2
Z(n) = 2n       (n >= 0)
Z(n) = -2n - 1  (n < 0)

对应 Python：

1
2
3
4
5
6
7
8
9
10
11
def zigzag_encode(n: int) -> int:
    return 2 * n if n >= 0 else -2 * n - 1


def zigzag_decode(n: int) -> int:
    return n // 2 if n % 2 == 0 else -(n // 2) - 1


for n in (-3, -2, -1, 0, 1, 2, 3):
    z = zigzag_encode(n)
    print(n, "->", z, "->", zigzag_decode(z))

在 32 位有符号整数范围内，常见位运算公式：

1
zigzag = (n << 1) ^ (n >> 31)

现在：

1
2
3
message Reading {
  sint32 temperature = 1;
}

当 temperature=-2 时：

1
2
3
4
5
ZigZag(-2) = 3
Varint(3) = 03
Tag(1, VARINT) = 08

最终：08 03

只需 2 字节（包含 Tag）。

5.7 字符串和子消息：Length-delimited

例如：

1
2
3
4
message Person {
  int32 id = 1;
  string name = 2;
}

当 id=150、name="Tom" 时，一种标准编码为：

1
08 96 01 12 03 54 6F 6D

拆开：

1
2
08 | 96 01       → 字段1，整数150
12 | 03 | 54 6F 6D → 字段2，长度3，UTF-8字符串"Tom"

这里 03 是字节长度，不是字符数量。

中文 "你好" 的 UTF-8 长度为 6：

1
2
3
4
"你好" = E4 BD A0 E5 A5 BD

字段 2 编码：
12 06 E4 BD A0 E5 A5 BD

嵌套消息也通常使用 Wire Type 2。它把子消息的字节作为 Payload，再加上长度前缀。

5.8 packed repeated、map 与 oneof 的编码

1
2
3
message Example {
  repeated int32 numbers = 1;
}

numbers = [1, 2, 3] 在 proto3 默认的 packed 编码下可表示为：

1
2
3
4
0A 03 01 02 03
│  │  └─────── 三个 Varint 数值
│  └────────── 长度 3
└───────────── 字段1，Wire Type 2

其他结构：

  • repeated string：通常是多条带同一字段编号的长度分隔记录。
  • map<K, V>：Wire Format 上相当于重复的 map-entry 子消息。
  • oneof：每个成员按自身字段类型编码；互斥关系来自 Schema 的解析与对象语义。
  • enum：通常按整数的 Varint 编码。

重要提醒：Protobuf 消息未必具有唯一的二进制字节序列；字段顺序、map 排序和未知字段处理等因素都可能影响实际 bytes。验证跨语言兼容时应比较消息语义和字段值，而不能一律要求 bytes 完全一致。

5.9 二进制手算综合题

Schema：

1
2
3
4
5
message Sensor {
  uint32 id = 1;
  sint32 temperature = 2;
  string unit = 3;
}

数据：

1
2
3
id = 300
temperature = -2
unit = "C"

计算：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
字段1：
  Tag = 08
  Varint(300) = AC 02
  → 08 AC 02

字段2：
  Tag = 10
  ZigZag(-2) = 3 → 03
  → 10 03

字段3：
  Tag = 1A
  长度 = 01
  UTF-8("C") = 43
  → 1A 01 43

拼接结果：

1
08 AC 02 10 03 1A 01 43

共 8 字节。

第 06 讲：Schema 演进、reserved 和未知字段

分布式系统通常无法保证所有服务和移动客户端同时升级。旧客户端可能还在读取新服务端产生的数据；历史缓存、日志和消息队列中也可能保存旧结构数据。

因此，Schema 不仅要“当前能用”，还要“长期可演进”。

6.1 向后兼容与向前兼容

假设：

V1：

1
2
3
4
message User {
  int32 id = 1;
  string name = 2;
}

V2：

1
2
3
4
5
message User {
  int32 id = 1;
  string name = 2;
  optional string email = 3;
}

定义：

  • 向后兼容（Backward Compatibility）：新程序能够读取旧数据（V2 读 V1）。
  • 向前兼容（Forward Compatibility）：旧程序能够读取新数据中的已知内容（V1 读 V2）。

这仅说明相关二进制数据能够正确解析，并不保证新增的业务校验、权限与处理逻辑也兼容。

6.2 修改字段的风险表

变更 二进制层面 主要风险
增加新编号字段 通常 Wire-safe 老程序不认识新字段
将字段 2 改为 5 Wire-unsafe / 语义破坏 原字段变未知字段
删除字段，不再复用编号 通常可逐步演进 历史数据、业务代码要处理
复用已发布的编号 危险 旧数据被解释成新含义
更改字段名但保留编号与类型 二进制通常可兼容 API、JSON、反射受到影响
增加枚举值 二进制通常可容纳 旧程序必须正确处理未知枚举值
int32 改为 int64 有条件兼容 新数值可能超出旧范围
int64 改为 string 通常不兼容 Wire Type 与语义不同
修改 oneof 结构 需单独评估 字段互斥、清除、未知状态问题

不同变更可能是 Wire-safe、Wire-compatible（有条件安全）或 Wire-unsafe。 JSON 的兼容性规则不等同于二进制格式。

6.3 字段编号永远不要随意复用

旧版：

1
2
3
4
5
message Payment {
  int64 id = 1;
  int64 amount_cents = 2;
  string card_number = 3;
}

错误演进：

1
2
3
4
5
message Payment {
  int64 id = 1;
  int64 amount_cents = 2;
  string note = 3; // 危险：复用了旧 card_number 的编号！
}

旧日志中的字段 3 可能被新程序解释成 note。这有时不仅是数据问题，还可能带来敏感信息暴露。

正确方式：

1
2
3
4
5
6
7
8
message Payment {
  reserved 3;
  reserved "card_number";

  int64 id = 1;
  int64 amount_cents = 2;
  string note = 4;
}

其他 reserved 写法：

1
2
3
reserved 3, 5, 8;
reserved 10 to 20;
reserved "old_name", "old_email";

不能混写：

1
2
// 错误：
reserved 3, "old_name";

reserved 主要在 Schema 编译期阻止编号或名称被重新使用。它不会神奇地删掉历史二进制文件里的旧字段。

6.4 Unknown Fields：向前兼容的关键

当 V1 收到 V2 的 email = 3 时：

  • V1 可以读取 id、name。
  • V1 的生成 API 没有 email 属性。
  • 现代 Protobuf 二进制解析通常会保留未知字段。
  • 如果 V1 直接重新序列化消息，V2 往往还能恢复 email。

但逐字段手动复制到一个新对象或转 JSON 再还原会导致未知字段丢失。

简单记忆：

1
2
V2 数据 → V1 解析 → V1 直接序列化 → V2 解析：未知字段通常保留
V2 数据 → V1 解析 → 手动复制已知字段 → V2 解析：未知字段丢失

6.5 完整 Python 实验：V2 → V1 → V2

目录如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
compatibility/
├── v1/
│   ├── order.proto
│   ├── order_pb2.py     # 生成
│   └── relay.py
├── v2/
│   ├── order.proto
│   ├── order_pb2.py     # 生成
│   ├── make.py
│   └── check.py
├── order_v2.bin         # 生成
├── forwarded.bin        # 生成
└── dropped.bin          # 生成

V1 的 v1/order.proto：

1
2
3
4
5
6
7
8
syntax = "proto3";
package shop;

message Order {
  int64 id = 1;
  int64 amount_cents = 2;
  string note = 3;
}

V2 的 v2/order.proto：

1
2
3
4
5
6
7
8
9
10
11
12
syntax = "proto3";
package shop;

message Order {
  reserved 3;
  reserved "note";

  int64 id = 1;
  int64 amount_cents = 2;
  optional string coupon_code = 4;
  optional bool is_paid = 5;
}

在 compatibility/ 目录执行：

1
2
python -m grpc_tools.protoc -I v1 --python_out=v1 v1/order.proto
python -m grpc_tools.protoc -I v2 --python_out=v2 v2/order.proto

为什么分两个目录并用独立进程运行？ 两份 Python 代码都定义了相同的完整 Protobuf 消息名 shop.Order，在同一个解释器中一起加载可能触发 Descriptor 冲突。独立进程最适合这个教学实验。

v2/make.py：使用 V2 创建消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from pathlib import Path
from order_pb2 import Order

root = Path(__file__).resolve().parents[1]

order = Order(
    id=10001,
    amount_cents=8999,
    coupon_code="SAVE20",
    is_paid=False,
)

assert order.HasField("coupon_code")
assert order.HasField("is_paid")

(root / "order_v2.bin").write_bytes(
    order.SerializeToString()
)
print("V2 写入完成")

v1/relay.py：旧版读取并分别保留/丢弃未知字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from pathlib import Path
from order_pb2 import Order

root = Path(__file__).resolve().parents[1]
data = (root / "order_v2.bin").read_bytes()

old_order = Order.FromString(data)

print("旧版可读取：", old_order.id, old_order.amount_cents)

# 路径 A：直接转发，保留未知字段
(root / "forwarded.bin").write_bytes(
    old_order.SerializeToString()
)

# 路径 B：主动丢弃未知字段
old_order.DiscardUnknownFields()

(root / "dropped.bin").write_bytes(
    old_order.SerializeToString()
)

v2/check.py：新版再次解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pathlib import Path
from order_pb2 import Order

root = Path(__file__).resolve().parents[1]

for filename in ("forwarded.bin", "dropped.bin"):
    data = (root / filename).read_bytes()
    order = Order.FromString(data)

    print(filename)
    print("  coupon:", repr(order.coupon_code))
    print("  coupon exists:", order.HasField("coupon_code"))
    print("  is_paid:", order.is_paid)
    print("  is_paid exists:", order.HasField("is_paid"))

运行顺序（从 compatibility/ 目录）：

1
2
3
python v2/make.py
python v1/relay.py
python v2/check.py

核心结果：

1
2
3
4
5
6
7
8
9
10
11
forwarded.bin
  coupon: 'SAVE20'
  coupon exists: True
  is_paid: False
  is_paid exists: True

dropped.bin
  coupon: ''
  coupon exists: False
  is_paid: False
  is_paid exists: False

这里尤其值得注意：

1
is_paid=False

虽然是布尔默认值，但因为字段是 optional，V2 显式设置它后 HasField() 为 True；当未知字段被丢弃后，它才变成未设置状态。

6.6 V3 挑战：为什么不能将 int64 改成 string？

错误变更：

1
2
3
4
5
// V1
int64 amount_cents = 2;

// V3
string amount_cents = 2;

原先的金额 300：

1
10 AC 02

改为字符串 "300"：

1
12 03 33 30 30

因为字段编号虽然同为 2，但 Wire Type 已从 0 (VARINT) 变为 2 (LEN)。旧程序可能把新记录当作未知字段，而非恢复金额。

更安全的思路：

1
2
3
4
5
6
7
8
9
10
message Order {
  reserved 3;
  reserved "note";

  int64 id = 1;
  int64 amount_cents = 2;           // 保留旧字段
  optional string coupon_code = 4;
  optional bool is_paid = 5;
  optional string amount_display = 6; // 另增显示用途字段
}

6.7 版本升级的实战流程

  1. 新增新编号字段，先保证未设置时的兼容行为。
  2. 让读取方逐步支持新字段，在旧客户端尚存时不要强行依赖它。
  3. 逐步升级写入方，必要时双写旧字段与新字段并定义权威来源。
  4. 验证历史数据、回滚、消息队列和服务混部。
  5. 退役旧字段后保留编号与名称，不要随意复用。

对于删除、类型调整、oneof 重构，不存在一个无条件适用的升级顺序。先写兼容矩阵和迁移方案，再进行灰度部署。

第 07 讲：Python 与 Java 跨语言交换

这一讲使用附件中的订单模型，以 .bin 文件为媒介，验证 Python 和 Java 能读取相同协议的数据。网络调用在第八讲引入。

7.1 项目目录与共享 Schema

1
2
3
4
5
6
7
8
9
10
protobuf_lesson7/
├── requirements.txt
├── python/exchange_app.py
├── java/
│   ├── pom.xml
│   └── src/main/
│       ├── proto/exchange.proto
│       └── java/com/example/exchange/ExchangeApp.java
├── compatibility/              # 第六讲实验
└── shared/                     # 运行时产生的 .bin 文件

共享文件是 java/src/main/proto/exchange.proto：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
syntax = "proto3";

package tutorial.exchange;

option java_package = "com.example.exchange.proto";
option java_multiple_files = true;

enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;
  ORDER_STATUS_PENDING = 1;
  ORDER_STATUS_PAID = 2;
}

message Product {
  int64 id = 1;
  string name = 2;
  int64 price_cents = 3;
}

message Order {
  int64 id = 1;
  OrderStatus status = 2;
  repeated Product products = 3;
  map<string, int32> quantities = 4;
  optional string coupon_code = 5;

  oneof contact {
    string email = 6;
    string phone = 7;
  }
}

package tutorial.exchange 是 Protobuf 命名空间，java_package 控制 Java 包名。Python 模块名来自文件名，生成结果为 exchange_pb2.py。

这个 Order 是独立示例，不是第四讲 ecommerce.Order 的升级版。实际交换双方需要使用对应的 Schema。

7.2 生成代码

在 protobuf_lesson7/ 根目录执行：

1
2
3
4
5
6
# 已具备依赖时跳过安装
python -m pip install -r requirements.txt
python -m grpc_tools.protoc -I java/src/main/proto --python_out=python java/src/main/proto/exchange.proto
cd java
mvn clean compile
cd ..

Maven 会生成 Java 消息类并编译程序。完整依赖和插件配置见下载项目的 java/pom.xml。已有独立 protoc 时，Python 消息生成命令也可以写成：

1
protoc -I java/src/main/proto --python_out=python java/src/main/proto/exchange.proto

7.3 Python 创建、序列化与解析订单

python/exchange_app.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import sys
from pathlib import Path
from exchange_pb2 import Order, Product, ORDER_STATUS_PAID


def create_order():
    order = Order(id=10001, status=ORDER_STATUS_PAID, coupon_code="SAVE20")
    order.products.add(id=101, name="Python Guide", price_cents=2999)
    order.products.add(id=102, name="Java Guide", price_cents=5999)
    order.quantities["101"] = 2
    order.quantities["102"] = 1
    order.email = "student@example.com"
    return order


def inspect_order(order):
    print("id:", order.id)
    print("status:", order.status)
    print("products:", [(p.id, p.name, p.price_cents) for p in order.products])
    print("quantities:", dict(order.quantities))
    print("coupon:", order.coupon_code, "present:", order.HasField("coupon_code"))
    print("contact:", order.WhichOneof("contact"), order.email)
    assert order.id == 10001
    assert len(order.products) == 2
    assert order.quantities["101"] == 2
    assert order.HasField("coupon_code")
    assert order.WhichOneof("contact") == "email"


def main():
    if len(sys.argv) != 3 or sys.argv[1] not in {"send", "read"}:
        raise SystemExit("Usage: python python/exchange_app.py send|read FILE")
    action, filename = sys.argv[1:]
    file = Path(filename)
    if action == "send":
        order = create_order()
        file.parent.mkdir(parents=True, exist_ok=True)
        file.write_bytes(order.SerializeToString())
        print("Python sent", file, "bytes:", file.stat().st_size)
        inspect_order(order)
    else:
        order = Order.FromString(file.read_bytes())
        print("Python received", file)
        inspect_order(order)


if __name__ == "__main__":
    main()

这里把商品与数量分开存放，是为了练习 repeated 和 map。正式订单可使用 OrderItem，把商品、成交价和购买数量放在同一行中。

7.4 Java 消息与 Builder

Java 消息不可直接修改，通过 Builder 创建或基于旧消息更新。完整代码见下载包中的 ExchangeApp.java。

操作 Python Java
创建消息 Order() Order.newBuilder().build()
读取 ID order.id order.getId()
添加子消息 order.products.add(...) builder.addProducts(...)
获取商品列表 order.products order.getProductsList()
获取 map dict(order.quantities) order.getQuantitiesMap()
optional 存在性 HasField() hasCouponCode()
oneof 当前成员 WhichOneof() getContactCase()
序列化 SerializeToString() toByteArray()
反序列化 Order.FromString(raw) Order.parseFrom(raw)
基于旧消息修改 赋值或 CopyFrom() order.toBuilder()

7.5 Python 写入，Java 读取

从项目根目录执行：

1
2
3
4
python python/exchange_app.py send shared/python_to_java.bin
cd java
mvn exec:java '-Dexec.args=read ../shared/python_to_java.bin'
cd ..

发送程序会创建 shared/。读取端会打印字段，并检查订单 ID、商品数量、购买数量、优惠码存在性和联系方式。

7.6 Java 写入，Python 读取

从项目根目录执行：

1
2
3
4
cd java
mvn exec:java '-Dexec.args=send ../shared/java_to_python.bin'
cd ..
python python/exchange_app.py read shared/java_to_python.bin

两个方向都应读出以下关键值，map 显示顺序可能不同：

1
2
3
4
5
订单 ID：10001
商品数：2
商品 101 的数量：2
优惠码：SAVE20，已设置
联系方式：email

跨语言验证主要比较字段值和存在性，而不是一律要求 bytes 完全一致。map 顺序与未知字段等都可能影响实际字节。

7.7 运行附件中的兼容性实验

在 protobuf_lesson7/ 根目录执行：

1
2
3
4
5
python -m grpc_tools.protoc -I compatibility/v1 --python_out=compatibility/v1 compatibility/v1/order.proto
python -m grpc_tools.protoc -I compatibility/v2 --python_out=compatibility/v2 compatibility/v2/order.proto
python compatibility/v2/make.py
python compatibility/v1/relay.py
python compatibility/v2/check.py

对照第六讲观察：V1 直接重新序列化后，V2 的未知字段通常仍然存在；主动清除后，优惠码和付款状态的存在性会消失。两个版本在独立进程运行，避免同名 Descriptor 冲突。

7.8 直接使用 TCP 时要注意什么？

Protobuf 负责消息编码，TCP 接收方还需要知道消息边界。可约定长度前缀：先读满 4 字节长度，再读满相应大小的 Payload，最后调用解析 API。一次 recv() 不保证返回全部数据，还要处理消息大小上限、断连和部分读取。gRPC 会提供消息分帧和远程调用机制。

第 08 讲：gRPC Python Server、Python/Java Client

前面通过 .bin 文件完成跨语言数据交换；现在升级为真正的网络服务：

1
2
3
4
5
6
7
8
9
10
11
             同一份 student_service.proto
                    ↓
           Protobuf 消息 + gRPC 服务
                    ↓
     ┌──────────────┴──────────────┐
     ↓                             ↓
Python Client                  Java Client
     └──────────────┬──────────────┘
                    ↓
         Python gRPC Server
           127.0.0.1:50051

8.1 RPC 与 Protobuf 的分工

  • RPC：Remote Procedure Call，远程过程调用。
  • Protobuf：定义请求/响应的数据结构，并对消息序列化。
  • gRPC：提供远程调用的 Stub、服务端框架、HTTP/2 传输、状态码、Deadline、流式通信等机制。

这意味着客户端可以写出类似调用本地方法的代码：

1
2
3
4
result = stub.GetStudent(
    pb.GetStudentRequest(id=1001),
    timeout=5,
)

而实际发生的是请求序列化、网络发送、服务端执行、响应序列化和客户端解析。

8.2 .proto 的 service 与 rpc

除了 message，还可以声明：

1
2
3
service StudentService {
  rpc GetStudent(GetStudentRequest) returns (Student);
}
  • service：远程服务。
  • rpc：一个远程方法。
  • 括号里的类型：请求消息。
  • returns (...)：响应消息。
  • 请求与响应必须是 Protobuf 消息类型（可以为空消息）。

8.3 gRPC 的四种通信模式

模式 请求 响应 .proto 写法
Unary 1 条 1 条 rpc Get(Req) returns (Resp);
Server Streaming 1 条 多条 rpc List(Req) returns (stream Resp);
Client Streaming 多条 1 条 rpc Upload(stream Req) returns (Resp);
Bidirectional Streaming 多条 多条 rpc Chat(stream Req) returns (stream Resp);

stream 表示 RPC 级别的消息流，不等同于单个 message 里面包含 repeated 列表。

本讲实现四个 RPC，覆盖两种通信模式：创建、查询和更新使用 Unary RPC，列表使用 Server Streaming。Client Streaming 与双向流留作扩展练习。

8.4 完整项目目录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
protobuf_lesson8/
├── README.md
├── requirements.txt
├── proto/
│   └── student_service.proto
├── python/
│   ├── server.py
│   ├── client.py
│   ├── smoke_test.py
│   ├── student_service_pb2.py        # 生成
│   └── student_service_pb2_grpc.py   # 生成
└── java/
    ├── pom.xml
    ├── target/generated-sources/     # 构建时生成
    └── src/main/java/com/example/school/
        └── StudentClient.java

环境：Python 3.10+、JDK 17+、Maven 3.8+、本机空闲端口 50051。

为了易于复现，完整项目的示例依赖版本放在 requirements.txt 和 pom.xml。构建前请确保 Python 与 Java 生成器及各自运行时版本兼容。

8.5 完整 proto/student_service.proto

该服务延续学生成绩系统的 Student 和 Score 消息。GetStudent 查询学生，CreateStudent 创建学生，UpdateScore 修改成绩，ListStudents 使用服务端流逐个返回学生。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
syntax = "proto3";
package tutorial.school;

option java_package = "com.example.school.proto";
option java_multiple_files = true;

enum Grade {
  GRADE_UNSPECIFIED = 0;
  GRADE_FIRST = 1;
  GRADE_SECOND = 2;
  GRADE_THIRD = 3;
  GRADE_FOURTH = 4;
}

message Score {
  string subject = 1;
  int32 points = 2;
}

message Student {
  int64 id = 1;
  string name = 2;
  optional int32 age = 3;
  Grade grade = 4;
  repeated Score scores = 5;

  oneof contact {
    string email = 6;
    string phone = 7;
  }
}

message GetStudentRequest {
  int64 id = 1;
}
message CreateStudentRequest {
  Student student = 1;
}
message ListStudentsRequest {}
message UpdateScoreRequest {
  int64 student_id = 1;
  string subject = 2;
  int32 points = 3;
}

service StudentService {
  rpc CreateStudent(CreateStudentRequest) returns (Student);
  rpc GetStudent(GetStudentRequest) returns (Student);
  rpc ListStudents(ListStudentsRequest) returns (stream Student);
  rpc UpdateScore(UpdateScoreRequest) returns (Student);
}

8.6 Python 依赖与代码生成

requirements.txt：

1
2
3
grpcio>=1.70,<2
grpcio-tools>=1.70,<2
protobuf>=6.30,<8

在项目根目录：

1
2
3
python -m pip install -r requirements.txt

python -m grpc_tools.protoc -I proto --python_out=python --grpc_python_out=python proto/student_service.proto

将生成：

1
2
python/student_service_pb2.py
python/student_service_pb2_grpc.py

两个文件的作用不同：

  • student_service_pb2.py：Student、Score、请求/响应消息、枚举等。
  • student_service_pb2_grpc.py：StudentServiceStub、StudentServiceServicer、注册服务端的方法。

如果不使用 Linux/macOS Shell，可以将多行生成命令写成单行：

1
python -m grpc_tools.protoc -I proto --python_out=python --grpc_python_out=python proto/student_service.proto

8.7 Python Server：完整实现

文件 python/server.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from concurrent import futures
import threading

import grpc
import student_service_pb2 as pb
import student_service_pb2_grpc as rpc

HOST = "127.0.0.1:50051"

class StudentService(rpc.StudentServiceServicer):
    def __init__(self):
        self.students = {}
        self.lock = threading.RLock()

    def CreateStudent(self, request, context):
        student = request.student
        if student.id <= 0 or not student.name.strip():
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "id must be positive and name nonblank")
        for score in student.scores:
            if not score.subject.strip() or not (0 <= score.points <= 100):
                context.abort(grpc.StatusCode.INVALID_ARGUMENT, "invalid subject or points")
        with self.lock:
            if student.id in self.students:
                context.abort(grpc.StatusCode.ALREADY_EXISTS, "student already exists")
            saved = pb.Student()
            saved.CopyFrom(student)
            self.students[student.id] = saved
            output = pb.Student()
            output.CopyFrom(saved)
            return output

    def GetStudent(self, request, context):
        with self.lock:
            student = self.students.get(request.id)
            if student is None:
                context.abort(grpc.StatusCode.NOT_FOUND, "student not found")
            output = pb.Student()
            output.CopyFrom(student)
            return output

    def ListStudents(self, request, context):
        with self.lock:
            snapshots = []
            for student in sorted(self.students.values(), key=lambda s: s.id):
                copied = pb.Student()
                copied.CopyFrom(student)
                snapshots.append(copied)
        for student in snapshots:
            yield student

    def UpdateScore(self, request, context):
        if not request.subject.strip() or not (0 <= request.points <= 100):
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "invalid subject or points")
        with self.lock:
            student = self.students.get(request.student_id)
            if student is None:
                context.abort(grpc.StatusCode.NOT_FOUND, "student not found")
            for score in student.scores:
                if score.subject == request.subject:
                    score.points = request.points
                    output = pb.Student()
                    output.CopyFrom(student)
                    return output
            context.abort(grpc.StatusCode.NOT_FOUND, "subject not found")


def make_server():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=8))
    rpc.add_StudentServiceServicer_to_server(StudentService(), server)
    port = server.add_insecure_port(HOST)
    if port == 0:
        raise RuntimeError("could not bind " + HOST)
    return server


def main():
    server = make_server()
    server.start()
    print("Python gRPC server listening at", HOST, flush=True)
    try:
        server.wait_for_termination()
    except KeyboardInterrupt:
        server.stop(grace=2).wait()


if __name__ == "__main__":
    main()

重点理解：

  1. rpc.StudentServiceServicer 是生成的服务端接口基类。
  2. 每个 Python 方法对应 .proto 中一个 rpc。
  3. request 是解析好的请求消息；context 表示当前 RPC 上下文。
  4. context.abort(grpc.StatusCode.NOT_FOUND, "...") 返回规范的 gRPC 失败状态。
  5. ListStudents 使用 yield 逐条发送响应。
  6. 教学示例使用字典存储数据，所以服务重启后数据会丢失；RLock 保护同一进程内对数据的并发访问。
  7. CopyFrom() 使服务端返回消息时不暴露正在被修改的内部对象。

服务端只监听：

1
127.0.0.1:50051

这是本机回环地址，且采用明文 gRPC。它只适合本地实验；生产环境必须进一步配置 TLS、认证授权和网络访问控制。

8.8 Python Client：完整实现

文件 python/client.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import grpc
import student_service_pb2 as pb
import student_service_pb2_grpc as rpc

TARGET = "127.0.0.1:50051"

def main():
    with grpc.insecure_channel(TARGET) as channel:
        stub = rpc.StudentServiceStub(channel)
        student = pb.Student(
            id=1001, name="Alice", age=18,
            grade=pb.GRADE_SECOND,
            email="alice@example.com",
        )
        student.scores.add(subject="Math", points=95)
        student.scores.add(subject="English", points=88)
        student.scores.add(subject="Computer", points=99)

        try:
            created = stub.CreateStudent(
                pb.CreateStudentRequest(student=student), timeout=5
            )
        except grpc.RpcError as e:
            if e.code() != grpc.StatusCode.ALREADY_EXISTS:
                raise
            print("Student already exists; querying current record")
            created = stub.GetStudent(pb.GetStudentRequest(id=1001), timeout=5)
        print("Created / loaded:", created.name, created.id)

        fetched = stub.GetStudent(pb.GetStudentRequest(id=1001), timeout=5)
        print("GetStudent:", fetched.name, "age present:", fetched.HasField("age"))
        print("Scores:", [(s.subject, s.points) for s in fetched.scores])
        print("Average:", sum(s.points for s in fetched.scores) / len(fetched.scores))
        print("Contact:", fetched.WhichOneof("contact"))

        updated = stub.UpdateScore(
            pb.UpdateScoreRequest(student_id=1001, subject="Computer", points=100),
            timeout=5,
        )
        print("Updated Computer score:", [
            s.points for s in updated.scores if s.subject == "Computer"
        ])

        print("ListStudents (server streaming):")
        for item in stub.ListStudents(pb.ListStudentsRequest(), timeout=5):
            print(" ", item.id, item.name)

        try:
            stub.GetStudent(pb.GetStudentRequest(id=-999), timeout=5)
        except grpc.RpcError as e:
            print("Expected missing student:", e.code().name)


if __name__ == "__main__":
    main()

这里的 StudentServiceStub(channel) 是客户端代理，通过它调用：

1
2
3
4
stub.CreateStudent(...)
stub.GetStudent(...)
stub.UpdateScore(...)
stub.ListStudents(...)

timeout=5 为每个 RPC 指定大约 5 秒的调用截止时间预算，以避免网络故障导致无限期等待。

对于服务端流：

1
2
for student in stub.ListStudents(pb.ListStudentsRequest(), timeout=5):
    print(student.id, student.name)

服务器不是返回一个 repeated Student 包装消息，而是在同一个 RPC 中陆续返回多条独立 Student 消息。

8.9 Java Maven 配置

Java 需要生成两类代码：

  1. 标准 Protobuf Java 消息类。
  2. gRPC Java 的 StudentServiceGrpc Stub 相关类。

文件 java/pom.xml：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>student-grpc-client</artifactId>
  <version>1.0-SNAPSHOT</version>
  <properties>
    <maven.compiler.release>17</maven.compiler.release>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <grpc.version>1.85.0</grpc.version>
    <protobuf.version>4.33.0</protobuf.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>io.grpc</groupId><artifactId>grpc-netty-shaded</artifactId>
      <version>${grpc.version}</version><scope>runtime</scope>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId><artifactId>grpc-protobuf</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId><artifactId>grpc-stub</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>com.google.protobuf</groupId><artifactId>protobuf-java</artifactId>
      <version>${protobuf.version}</version>
    </dependency>
    <dependency>
      <groupId>javax.annotation</groupId><artifactId>javax.annotation-api</artifactId>
      <version>1.3.2</version>
    </dependency>
  </dependencies>
  <build>
    <extensions>
      <extension>
        <groupId>kr.motd.maven</groupId><artifactId>os-maven-plugin</artifactId>
        <version>1.7.1</version>
      </extension>
    </extensions>
    <plugins>
      <plugin>
        <groupId>org.xolstice.maven.plugins</groupId>
        <artifactId>protobuf-maven-plugin</artifactId>
        <version>0.6.1</version>
        <configuration>
          <protoSourceRoot>${project.basedir}/../proto</protoSourceRoot>
          <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
          <pluginId>grpc-java</pluginId>
          <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
        </configuration>
        <executions>
          <execution><goals><goal>compile</goal><goal>compile-custom</goal></goals></execution>
        </executions>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.13.0</version>
      </plugin>
      <plugin>
        <groupId>org.codehaus.mojo</groupId>
        <artifactId>exec-maven-plugin</artifactId>
        <version>3.5.0</version>
        <configuration>
          <mainClass>com.example.school.StudentClient</mainClass>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

Maven 插件使用共享的 ../proto 文件夹，并在 mvn clean compile 时运行 protoc 与 protoc-gen-grpc-java。

8.10 Java Client：完整实现

Java 用 ManagedChannelBuilder 建立 Channel，使用 StudentServiceGrpc.newBlockingStub(channel) 获取阻塞式 Stub，通过 Java Builder 构造请求：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
package com.example.school;

import com.example.school.proto.CreateStudentRequest;
import com.example.school.proto.GetStudentRequest;
import com.example.school.proto.Grade;
import com.example.school.proto.ListStudentsRequest;
import com.example.school.proto.Score;
import com.example.school.proto.Student;
import com.example.school.proto.StudentServiceGrpc;
import com.example.school.proto.UpdateScoreRequest;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;

import java.util.concurrent.TimeUnit;

public final class StudentClient {
    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder
                .forAddress("127.0.0.1", 50051)
                .usePlaintext() // local tutorial only; production should use TLS
                .build();

        try {
            StudentServiceGrpc.StudentServiceBlockingStub stub =
                    StudentServiceGrpc.newBlockingStub(channel);

            Student student = Student.newBuilder()
                    .setId(2002)
                    .setName("Bob")
                    .setAge(19)
                    .setGrade(Grade.GRADE_SECOND)
                    .addScores(Score.newBuilder().setSubject("Math").setPoints(93).build())
                    .addScores(Score.newBuilder().setSubject("English").setPoints(91).build())
                    .setEmail("bob@example.com")
                    .build();

            try {
                Student created = stub.withDeadlineAfter(5, TimeUnit.SECONDS).createStudent(CreateStudentRequest.newBuilder()
                        .setStudent(student).build());
                System.out.println("Created: " + created.getId() + " " + created.getName());
            } catch (StatusRuntimeException e) {
                if (e.getStatus().getCode() != Status.Code.ALREADY_EXISTS) throw e;
                System.out.println("Student 2002 already exists");
            }

            Student fetched = stub.withDeadlineAfter(5, TimeUnit.SECONDS).getStudent(
                    GetStudentRequest.newBuilder().setId(2002).build());
            System.out.println("Fetched: " + fetched.getName()
                    + " age present=" + fetched.hasAge()
                    + " contact=" + fetched.getContactCase());
            for (Score score : fetched.getScoresList()) {
                System.out.println("  " + score.getSubject() + ": " + score.getPoints());
            }

            Student updated = stub.withDeadlineAfter(5, TimeUnit.SECONDS).updateScore(UpdateScoreRequest.newBuilder()
                    .setStudentId(2002)
                    .setSubject("Math")
                    .setPoints(100)
                    .build());
            System.out.println("Updated score: " + updated.getScores(0).getPoints());

            System.out.println("ListStudents (server streaming):");
            var iterator = stub.withDeadlineAfter(5, TimeUnit.SECONDS).listStudents(ListStudentsRequest.getDefaultInstance());
            while (iterator.hasNext()) {
                Student s = iterator.next();
                System.out.println("  " + s.getId() + " " + s.getName());
            }

            try {
                stub.withDeadlineAfter(5, TimeUnit.SECONDS).getStudent(GetStudentRequest.newBuilder().setId(-999).build());
                throw new IllegalStateException("expected NOT_FOUND");
            } catch (StatusRuntimeException e) {
                if (e.getStatus().getCode() != Status.Code.NOT_FOUND) throw e;
                System.out.println("Expected error: " + e.getStatus().getCode());
            }
        } finally {
            channel.shutdown();
            if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
                channel.shutdownNow();
            }
        }
    }
}

注意和 Python 的 API 差异：

操作 Python Java
建立 Channel grpc.insecure_channel() ManagedChannelBuilder.forAddress()
创建 Stub StudentServiceStub(...) StudentServiceGrpc.newBlockingStub(...)
查询 RPC stub.GetStudent(...) stub.getStudent(...)
设置超时 timeout=5 .withDeadlineAfter(5, TimeUnit.SECONDS)
读取流 for item in stub.ListStudents(...) Iterator<Student>
错误 grpc.RpcError StatusRuntimeException

Java 的 withDeadlineAfter() 在每次 RPC 前设置新的截止时间。只在初始化 Stub 时设置一次并长期复用，会让后续调用继承不断减少甚至已经过期的时间预算。Python 则在每次调用中传入 timeout=5。

Java 的 usePlaintext() 仅用于这个本机示例；对外提供服务时应改为安全传输，并补齐服务端身份与访问控制。

8.11 按顺序启动完整项目

终端 A：首次生成 Python 代码

在 protobuf_lesson8/ 根目录：

1
2
python -m pip install -r requirements.txt
python -m grpc_tools.protoc -I proto --python_out=python --grpc_python_out=python proto/student_service.proto

终端 A：启动 Python Server

1
python python/server.py

预期：

1
Python gRPC server listening at 127.0.0.1:50051

让这个进程保持运行。

终端 B：运行 Python Client

在项目根目录：

1
python python/client.py

核心预期输出：

1
2
3
4
5
6
7
8
9
Created / loaded: Alice 1001
GetStudent: Alice age present: True
Scores: [('Math', 95), ('English', 88), ('Computer', 99)]
Average: 94.0
Contact: email
Updated Computer score: [100]
ListStudents (server streaming):
  1001 Alice
Expected missing student: NOT_FOUND

终端 C：构建并运行 Java Client

在项目根目录：

1
2
3
cd java
mvn clean compile
mvn exec:java

Java 会创建 ID 为 2002 的 Bob，调用查询、修改、服务端流式列表，并演示 NOT_FOUND。

如果此前已运行 Python Client，Java 列表中应能看到 Alice 与 Bob；如果只运行 Java Client，则只有 Bob。重启 Python Server 后内存数据库会清空。

8.12 自动化 Smoke Test

项目还包含一个独立的 Python 自动测试脚本，用于验证创建、查询、更新、流式返回和错误码：

文件 python/smoke_test.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
"""Run from project root after generating Python protobuf files:
python python/smoke_test.py
No separately running server required.
"""
import grpc
import student_service_pb2 as pb
import student_service_pb2_grpc as rpc
from server import make_server

def main():
    server = make_server()
    server.start()
    try:
        with grpc.insecure_channel("127.0.0.1:50051") as channel:
            stub = rpc.StudentServiceStub(channel)
            student = pb.Student(id=1001, name="Alice", age=18, grade=pb.GRADE_SECOND)
            student.scores.add(subject="Math", points=95)
            created = stub.CreateStudent(pb.CreateStudentRequest(student=student), timeout=5)
            assert created.HasField("age") and created.age == 18
            fetched = stub.GetStudent(pb.GetStudentRequest(id=1001), timeout=5)
            assert fetched.name == "Alice"
            result = stub.UpdateScore(pb.UpdateScoreRequest(
                student_id=1001, subject="Math", points=100), timeout=5)
            assert result.scores[0].points == 100
            assert len(list(stub.ListStudents(pb.ListStudentsRequest(), timeout=5))) == 1
            try:
                stub.GetStudent(pb.GetStudentRequest(id=999), timeout=5)
                raise AssertionError("NOT_FOUND expected")
            except grpc.RpcError as e:
                assert e.code() == grpc.StatusCode.NOT_FOUND
            print("PASS: Python gRPC unary, server-streaming and error handling")
    finally:
        server.stop(grace=1).wait()

if __name__ == "__main__":
    main()

在**没有另行启动 python/server.py**的情况下运行：

1
python python/smoke_test.py

如果通过，会打印：

1
PASS: Python gRPC unary, server-streaming and error handling

此测试负责启动临时服务端，因此不能与占用同一个端口的主服务端同时运行。

8.13 gRPC 状态码与异常处理

状态码 用途
OK 调用成功
INVALID_ARGUMENT 输入参数非法
NOT_FOUND 学生/课程不存在
ALREADY_EXISTS 重复创建学生
FAILED_PRECONDITION 当前状态不允许执行操作
UNAVAILABLE 服务临时不可用
DEADLINE_EXCEEDED RPC 截止时间已到
UNAUTHENTICATED 缺少有效认证
PERMISSION_DENIED 没有权限

Python：

1
2
3
4
5
6
7
8
try:
    student = stub.GetStudent(
        pb.GetStudentRequest(id=9999),
        timeout=5,
    )
except grpc.RpcError as e:
    print(e.code())
    print(e.details())

Java：

1
2
3
4
5
6
try {
    Student student = stub.withDeadlineAfter(5, TimeUnit.SECONDS).getStudent(
        GetStudentRequest.newBuilder().setId(9999).build());
} catch (io.grpc.StatusRuntimeException e) {
    System.out.println(e.getStatus().getCode());
}

别把所有错误都当成可以重试：例如 INVALID_ARGUMENT 不应该机械重试，而某些 UNAVAILABLE 可以配合退避策略重试。创建订单、扣减库存等非幂等操作的重试需要额外设计去重与一致性机制。

8.14 gRPC 进阶：生产化必须补上的能力

本章是教学版网络服务，不是开箱即用的生产架构。要上线时还应考虑：

  • 传输安全：TLS / mTLS、服务端身份校验。
  • 业务安全：认证、授权、细粒度资源权限。
  • 可靠性：Deadline、优雅关闭、重试策略、幂等性。
  • 运行状态：健康检查、指标、日志、链路追踪。
  • 维护性：Proto Schema 版本治理、兼容性测试、自动生成代码。
  • 规模化：负载均衡、服务发现、流控、容量与背压。
  • 数据层：数据库、事务和持久化；不要把样例内存字典直接当生产数据库。

回顾与进一步练习

八讲串起了 Schema 设计、代码生成、二进制读写、版本演进、跨语言交换和网络服务。可以用下面的问题自查：

  1. 为什么字段编号不能随意修改或复用？
  2. optional 怎样区分未设置和显式设置默认值？
  3. 08 96 01 对应哪个字段、哪种 Wire Type、什么值？
  4. 为什么 sint32 适合绝对值较小的负数？
  5. 旧程序直接转发和逐字段重建消息有什么区别？
  6. repeated Student 和服务端流有什么不同？
  7. Java 的 Deadline 为什么应在每次调用前设置？

可以继续为学生服务增加平均分查询：先定义请求、响应和 RPC，再重新生成代码，补齐服务端和客户端。成绩 95、88、99 的平均分为 94.0；如果已经将 Computer 改为 100，平均分则约为 94.3333。验证时要注意状态和运行顺序。

官方参考资料

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

一个热爱游戏的普通人<br>努力敲代码ing