Python消息序列化用Msgpack吗

本文目录导读：

1. 主要特点
2. 基本使用
3. 与 JSON 对比
4. 高级特性
5. 适用场景
6. 注意事项

是的，Msgpack（MessagePack） 是 Python 中一种高效的消息序列化格式，通常用于替代 JSON 或 Protocol Buffers。

主要特点

• 紧凑的二进制格式：比 JSON 更小、更快
• 跨语言支持：多种编程语言都能解析
• 支持复杂数据类型：列表、字典、字节串等

基本使用

安装

pip install msgpack

序列化与反序列化

import msgpack
# 序列化
data = {"name": "Alice", "age": 30, "scores": [85, 92, 78]}
packed = msgpack.packb(data)  # 返回 bytes
# 反序列化
unpacked = msgpack.unpackb(packed)
print(unpacked)  # {'name': 'Alice', 'age': 30, 'scores': [85, 92, 78]}

文件操作

# 写入文件
with open('data.msgpack', 'wb') as f:
    msgpack.pack(data, f)
# 读取文件
with open('data.msgpack', 'rb') as f:
    data = msgpack.unpack(f)

与 JSON 对比

import json
import msgpack
# JSON
json_data = json.dumps({"hello": "world"})
print(len(json_data))  # 16 bytes
# MessagePack
msgpack_data = msgpack.packb({"hello": "world"})
print(len(msgpack_data))  # 12 bytes（更紧凑）

高级特性

自定义类型序列化

import msgpack
from datetime import datetime
# 定义自定义编码器
def encode_custom(obj):
    if isinstance(obj, datetime):
        return {'__datetime__': True, 'value': obj.isoformat()}
    return obj
def decode_custom(obj):
    if '__datetime__' in obj:
        return datetime.fromisoformat(obj['value'])
    return obj
# 使用自定义编码
data = {"time": datetime.now()}
packed = msgpack.packb(data, default=encode_custom)
unpacked = msgpack.unpackb(packed, object_hook=decode_custom)

流式处理

import msgpack
# 流式打包多个对象
with open('stream.msgpack', 'wb') as f:
    for item in items:
        msgpack.pack(item, f)
# 流式解包
with open('stream.msgpack', 'rb') as f:
    unpacker = msgpack.Unpacker(f)
    for item in unpacker:
        print(item)

适用场景

• 高性能网络通信：如 RPC、消息队列
• 存储紧凑数据：减少磁盘/内存占用
• 跨语言数据传输：Python↔️Java↔️Go↔️C++等
• 嵌入式系统：资源受限环境

注意事项

1. 非直接人类可读：二进制格式，调试不如 JSON 方便
2. 无类型校验：不像 Protocol Buffers 有强类型约束
3. 安全性：反序列化不可信数据时建议用 raw=False 避免代码执行风险