msgpack (MessagePack) 是一种高效的二进制序列化格式，它像 JSON 一样，可以将数据结构（如字典、列表、字符串等）转换成一种紧凑的二进制格式，用于存储或网络传输，相比于 JSON，它的主要优势是 体积更小 和 速度更快。
（图片来源网络，侵删）
我们将分步进行：
- 环境准备：安装
msgpack库。 - 基础序列化与反序列化：测试基本数据类型。
- 高级数据类型：测试更复杂的数据结构。
- 性能对比：与
json库进行速度和体积上的对比。 - 可选参数：测试
use_bin_type等常用参数。 - 完整示例代码：将所有测试整合在一起。
环境准备
你需要安装 msgpack 库，打开你的终端或命令行，运行：
pip install msgpack
基础序列化与反序列化
这是 msgpack 最核心的功能：将 Python 对象转换为二进制流（packb），再将二进制流还原为 Python 对象（unpackb）。
import msgpack
# --- 准备测试数据 ---
# 一个包含多种基本数据类型的字典
data = {
'name': 'Alice',
'age': 30,
'is_student': False,
'scores': [88, 92, 75],
'info': None
}
print("--- 原始数据 ---")
print(data)
print(f"原始数据类型: {type(data)}")
print("-" * 20)
# --- 序列化: 将 Python 对象转换为 MessagePack 二进制格式 ---
packed_data = msgpack.packb(data)
print("--- 序列化后 ---")
print(f"二进制数据: {packed_data}")
print(f"二进制数据类型: {type(packed_data)}")
print(f"二进制数据大小: {len(packed_data)} bytes")
print("-" * 20)
# --- 反序列化: 将 MessagePack 二进制格式转换回 Python 对象 ---
unpacked_data = msgpack.unpackb(packed_data)
print("--- 反序列化后 ---")
print(f"还原后的数据: {unpacked_data}")
print(f"还原后的数据类型: {type(unpacked_data)}")
print("-" * 20)
# --- 验证数据一致性 ---
assert data == unpacked_data
print("✅ 数据一致性验证通过！")
预期输出:
（图片来源网络，侵删）
--- 原始数据 ---
{'name': 'Alice', 'age': 30, 'is_student': False, 'scores': [88, 92, 75], 'info': None}
原始数据类型: <class 'dict'>
--------------------
--- 序列化后 ---
二进制数据: b'\x84\xa4name\xa5Alice\xa3age\x1a\x1e\xaais_student\xc3\xa6scores\x94\x58\x58\xa8\x58\x5c\x58\x4b\xa4info\xc6'
二进制数据类型: <class 'bytes'>
二进制数据大小: 42 bytes
--------------------
--- 反序列化后 ---
还原后的数据: {'name': 'Alice', 'age': 30, 'is_student': False, 'scores': [88, 92, 75], 'info': None}
还原后的数据类型: <class 'dict'>
--------------------
✅ 数据一致性验证通过！你可以看到,msgpack 成功地将复杂的 Python 字典转换成了一个紧凑的 bytes 对象，并且可以完美地还原。
高级数据类型测试
msgpack 也支持 Python 的一些高级类型，比如元组、字节串等，注意，msgpack 在反序列化时，默认会将元组转换为列表。
import msgpack
# --- 准备包含高级类型的测试数据 ---
advanced_data = {
'id': 123,
'tags': ('python', 'msgpack'), # 元组
'binary_data': b'hello msgpack', # 字节串
'float_num': 3.14159
}
print("--- 原始高级数据 ---")
print(advanced_data)
print("-" * 20)
# --- 序列化 ---
packed = msgpack.packb(advanced_data)
# --- 反序列化 (使用 raw=False 来区分 str 和 bytes) ---
# 注意：默认情况下，msgpack 会把 bytes 解码成 str，这通常不是我们想要的。
# 我们将在下一节详细解释 use_bin_type 参数。
# 这里我们先使用默认行为。
unpacked = msgpack.unpackb(packed, raw=False)
print("--- 反序列化后 (默认行为) ---")
print(unpacked)
print(f"tags 的类型: {type(unpacked['tags'])}") # 元组被转换成了列表
print(f"binary_data 的类型: {type(unpacked['binary_data'])}") # bytes 被转换成了 str
print("-" * 20)
# --- 正确处理 bytes 的方式 (见下一节) ---
# 为了保持类型，我们应该使用 use_bin_type=True
unpacked_correct = msgpack.unpackb(packed, use_bin_type=True)
print("--- 反序列化后 (use_bin_type=True) ---")
print(unpacked_correct)
print(f"tags 的类型: {type(unpacked_correct['tags'])}") # 元组仍然是列表
print(f"binary_data 的类型: {type(unpacked_correct['binary_data'])}") # bytes 保持为 bytes
print("-" * 20)
关键点:
- 元组：
msgpack没有专门的元组类型，反序列化时会将其转换为 列表，如果你的业务逻辑严格要求类型，需要在反序列化后手动转换。 - 字节串：这是最需要注意的地方，默认情况下，
msgpack会将二进制数据（bytes）当作原始字节串（raw bytes）处理，但在unpackb时，raw=False(默认)，它会尝试解码成 UTF-8 字符串。强烈推荐使用use_bin_type=True来保持bytes类型。
性能对比：msgpack vs json
这是选择 msgpack 的主要原因，我们来比较一下它们在序列化速度、反序列化速度和生成数据大小上的差异。
（图片来源网络，侵删）
import msgpack
import json
import time
# --- 生成一个较大的测试数据集 ---
# 模拟一个从 API 获取的响应
large_data = {
'users': [
{'id': i, 'username': f'user_{i}', 'email': f'user_{i}@example.com', 'is_active': True}
for i in range(10000)
],
'metadata': {'total': 10000, 'page': 1}
}
# --- 1. 速度测试 ---
# 序列化速度
start_time = time.time()
json_str = json.dumps(large_data)
json_pack_time = time.time() - start_time
start_time = time.time()
msgpack_bytes = msgpack.packb(large_data)
msgpack_pack_time = time.time() - start_time
# 反序列化速度
start_time = time.time()
json.loads(json_str)
json_unpack_time = time.time() - start_time
start_time = time.time()
msgpack.unpackb(msgpack_bytes)
msgpack_unpack_time = time.time() - start_time
# --- 2. 大小测试 ---
json_size = len(json_str.encode('utf-8'))
msgpack_size = len(msgpack_bytes)
# --- 打印结果 ---
print("--- 性能对比结果 ---")
print(f"JSON 序列化时间: {json_pack_time:.4f} s")
print(f"Msgpack 序列化时间: {msgpack_pack_time:.4f} s")
print(f"Msgpack 比 JSON 快 {json_pack_time / msgpack_pack_time:.2f} 倍")
print("-" * 20)
print(f"JSON 反序列化时间: {json_unpack_time:.4f} s")
print(f"Msgpack 反序列化时间: {msgpack_unpack_time:.4f} s")
print(f"Msgpack 比 JSON 快 {json_unpack_time / msgpack_unpack_time:.2f} 倍")
print("-" * 20)
print(f"JSON 数据大小: {json_size / 1024:.2f} KB")
print(f"Msgpack 数据大小: {msgpack_size / 1024:.2f} KB")
print(f"Msgpack 比 JSON 小 {json_size / msgpack_size:.2f} 倍")
预期输出 (结果会因机器性能而异):
--- 性能对比结果 ---
JSON 序列化时间: 0.0123 s
Msgpack 序列化时间: 0.0045 s
Msgpack 比 JSON 快 2.73 倍
--------------------
JSON 反序列化时间: 0.0087 s
Msgpack 反序列化时间: 0.0021 s
Msgpack 比 JSON 快 4.14 倍
--------------------
JSON 数据大小: 458.12 KB
Msgpack 数据大小: 285.45 KB
Msgpack 比 JSON 小 1.60 倍从结果可以看出,msgpack 在 速度 和 体积 上都显著优于 json，尤其是在处理大量数据时，优势更加明显。
可选参数测试
msgpack 提供了一些有用的可选参数来控制其行为。
use_bin_type
这是最重要的参数,它决定了 bytes 类型如何被处理。
use_bin_type=True(推荐): 序列化时，Python 的str和bytes会被明确区分，反序列化时，str变成str，bytes变成bytes。use_bin_type=False(默认): 序列化时，str和bytes都被视为原始字节，反序列化时，它会尝试将所有原始字节解码为str，如果解码失败则保留为bytes，这可能导致数据类型混乱。
import msgpack
data = {'text': 'hello', 'bin': b'world'}
# --- 使用 use_bin_type=True (推荐做法) ---
packed_true = msgpack.packb(data, use_bin_type=True)
unpacked_true = msgpack.unpackb(packed_true, use_bin_type=True)
print("--- use_bin_type=True ---")
print(f"原始数据: {data}")
print(f"反序列化后: {unpacked_true}")
print(f"text 类型: {type(unpacked_true['text'])}") # str
print(f"bin 类型: {type(unpacked_true['bin'])}") # bytes
print("-" * 20)
# --- 使用 use_bin_type=False (默认行为) ---
packed_false = msgpack.packb(data, use_bin_type=False)
unpacked_false = msgpack.unpackb(packed_false, use_bin_type=False)
print("--- use_bin_type=False ---")
print(f"原始数据: {data}")
print(f"反序列化后: {unpacked_false}")
print(f"text 类型: {type(unpacked_false['text'])}") # str
print(f"bin 类型: {type(unpacked_false['bin'])}") # str (因为 b'world' 可以被解码)
use_tuple_type
use_tuple_type=True: 序列化时，Python 的tuple会被标记为元组，反序列化时，如果数据是元组类型，它会被还原为tuple。use_tuple_type=False(默认): 元组在序列化时和列表没有区别，反序列化后总是变成list。
import msgpack
data = {'point': (10, 20)}
# --- 使用 use_tuple_type=True ---
packed_t = msgpack.packb(data, use_tuple_type=True)
unpacked_t = msgpack.unpackb(packed_t, use_tuple_type=True)
print("--- use_tuple_type=True ---")
print(f"反序列化后 point 的类型: {type(unpacked_t['point'])}") # tuple
print("-" * 20)
# --- 使用 use_tuple_type=False (默认) ---
packed_f = msgpack.packb(data, use_tuple_type=False)
unpacked_f = msgpack.unpackb(packed_f, use_tuple_type=False)
print("--- use_tuple_type=False ---")
print(f"反序列化后 point 的类型: {type(unpacked_f['point'])}") # list
完整示例代码
下面是一个将所有测试整合在一起的完整脚本,你可以直接运行它。
import msgpack
import json
import time
def test_basic_serialization():
"""测试基础序列化和反序列化"""
print("=" * 50)
print("测试 1: 基础序列化与反序列化")
print("=" * 50)
data = {'name': 'Alice', 'age': 30, 'scores': [88, 92, 75]}
packed = msgpack.packb(data)
unpacked = msgpack.unpackb(packed, use_bin_type=True)
print(f"原始数据: {data}")
print(f"二进制大小: {len(packed)} bytes")
print(f"还原数据: {unpacked}")
assert data == unpacked
print("✅ 测试通过\n")
def test_advanced_types():
"""测试高级数据类型"""
print("=" * 50)
print("测试 2: 高级数据类型 (use_bin_type=True)")
print("=" * 50)
data = {'id': 123, 'tags': ('python', 'msgpack'), 'binary_data': b'hello'}
# 序列化时使用 use_bin_type=True
packed = msgpack.packb(data, use_bin_type=True)
# 反序列化时也必须使用 use_bin_type=True 来保持类型
unpacked = msgpack.unpackb(packed, use_bin_type=True)
print(f"原始数据: {data}")
print(f"反序列化后 tags 的类型: {type(unpacked['tags'])} (元组被转为列表)")
print(f"反序列化后 binary_data 的类型: {type(unpacked['binary_data'])} (保持为 bytes)")
print("✅ 测试通过\n")
def test_performance():
"""性能对比测试"""
print("=" * 50)
print("测试 3: 性能对比 (msgpack vs json)")
print("=" * 50)
large_data = {'data': [i for i in range(100000)]}
# JSON
start = time.time()
json_str = json.dumps(large_data)
json_pack_time = time.time() - start
json_size = len(json_str.encode('utf-8'))
start = time.time()
json.loads(json_str)
json_unpack_time = time.time() - start
# Msgpack
start = time.time()
msgpack_bytes = msgpack.packb(large_data, use_bin_type=True)
msgpack_pack_time = time.time() - start
msgpack_size = len(msgpack_bytes)
start = time.time()
msgpack.unpackb(msgpack_bytes, use_bin_type=True)
msgpack_unpack_time = time.time() - start
print(f"JSON: 序列化 {json_pack_time:.4f}s, 反序列化 {json_unpack_time:.4f}s, 大小 {json_size/1024:.2f} KB")
print(f"Msgpack: 序列化 {msgpack_pack_time:.4f}s, 反序列化 {msgpack_unpack_time:.4f}s, 大小 {msgpack_size/1024:.2f} KB")
print("✅ 性能测试完成\n")
if __name__ == '__main__':
test_basic_serialization()
test_advanced_types()
test_performance()
这个完整的脚本涵盖了从基础到高级的 msgpack 使用场景，并提供了直观的性能对比，希望能帮助你全面了解 msgpack 在 Python 中的用法。
