功能说明
LEB128 是一种变长整数编码:小数值只占一个字节,大数值按 7 位一组逐字节扩展,最高位表示是否还有后续字节。它广泛用于 Protocol Buffers、WebAssembly 和数据库索引等场景。这段代码实现无符号整数的编码、带偏移量的顺序解码,以及有符号数的 ZigZag 变换。
顺序解码很重要:一个字节流里可以连续放多个 varint,解码函数返回“值 + 下一个偏移量”,调用方据此继续读取,不需要长度前缀。
关键代码
encode_varint 每次取低 7 位,若还有剩余位就设置 0x80 继续位;decode_varint 反向累加 (byte & 0x7F) << shift,遇到最高位为 0 的字节结束,并检查截断、超长和超过 64 位的情况。
zigzag_encode 把有符号整数映射成无符号数:负数变成正奇数,非负数变成正偶数,这样绝对值小的负数也能用很少的字节表示。zigzag_decode 是其逆变换。
使用注意
编码不接受负数,负数必须先做 ZigZag 或使用其他有符号编码。解码器接受非规范编码(例如用两个字节表示 0),如果协议要求唯一表示,需要额外拒绝过长编码。偏移量必须指向完整 varint 的起始位置,否则会解析错误。
zigzag_encode 中的 >> 63 假定输入在 64 位范围内;Python 整数本身无上限,超出范围时结果不再符合标准 ZigZag。字节流解析还应结合总长度上限,避免恶意输入导致循环。
验证方式
示例对 0、1、127、128、300 和 2^32 编码后立即解码,断言值与偏移量都正确,并打印十六进制结果,例如 300 编码为 ac02。连续编码 0 和 300 后按偏移量解码,得到 (0, 300, 长度)。ZigZag 对 -500、-1、0、1、500 都能往返一致。
代码示例
from __future__ import annotations
LIMIT = 1 << 64
def encode_varint(value: int) -> bytes:
if not 0 <= value < LIMIT:
raise ValueError("value must fit in 64 bits")
output = bytearray()
while True:
byte = value & 0x7F
value >>= 7
if value:
output.append(byte | 0x80)
else:
output.append(byte)
return bytes(output)
def decode_varint(data: bytes, offset: int = 0) -> tuple[int, int]:
value = 0
shift = 0
while True:
if offset >= len(data):
raise ValueError("truncated varint")
byte = data[offset]
offset += 1
value |= (byte & 0x7F) << shift
if not byte & 0x80:
if value >= LIMIT:
raise ValueError("varint exceeds 64 bits")
return value, offset
shift += 7
if shift > 63:
raise ValueError("varint too long")
def zigzag_encode(value: int) -> int:
return (value << 1) ^ (value >> 63)
def zigzag_decode(value: int) -> int:
return (value >> 1) ^ -(value & 1)
def main() -> None:
for number in [0, 1, 127, 128, 300, 2**32]:
encoded = encode_varint(number)
decoded, next_offset = decode_varint(encoded)
print(number, encoded.hex(), decoded)
assert decoded == number
assert next_offset == len(encoded)
combined = encode_varint(0) + encode_varint(300)
first, offset = decode_varint(combined)
second, offset = decode_varint(combined, offset)
assert (first, second, offset) == (0, 300, len(combined))
for value in [-500, -1, 0, 1, 500]:
assert zigzag_decode(zigzag_encode(value)) == value
if __name__ == "__main__":
main()