当前位置：首页 > news >正文

Python序列化与反序列化：从JSON到高性能二进制格式

news 2026/5/10 9:38:50

Python序列化与反序列化：从JSON到高性能二进制格式

引言

序列化是将对象转换为可存储或传输格式的过程，反序列化则是其逆过程。在后端开发中，序列化广泛应用于数据持久化、网络通信、缓存等场景。

Python提供了多种序列化方案，从简单的JSON到高性能的二进制格式。本文将深入探讨各种序列化方案的原理、优缺点及适用场景。

一、Python标准库序列化方案

1.1 JSON序列化

JSON是最常用的序列化格式，具有良好的可读性和跨语言兼容性：

import json # 基本数据类型序列化 data = { 'name': 'John', 'age': 30, 'scores': [95, 87, 92], 'is_active': True, 'metadata': None } # 序列化 json_str = json.dumps(data, indent=2) print(json_str) # 反序列化 parsed_data = json.loads(json_str) print(parsed_data['name']) # 输出: John

1.2 JSON高级特性

from datetime import datetime import json # 自定义编码器处理datetime class DateTimeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() return super().default(obj) # 使用自定义编码器 data = { 'event': 'meeting', 'time': datetime.now(), 'attendees': ['Alice', 'Bob'] } json_str = json.dumps(data, cls=DateTimeEncoder) print(json_str) # 自定义解码器 def datetime_decoder(obj): if 'time' in obj: obj['time'] = datetime.fromisoformat(obj['time']) return obj parsed = json.loads(json_str, object_hook=datetime_decoder) print(parsed['time']) # datetime对象

1.3 pickle序列化

pickle是Python特有的序列化格式，支持几乎所有Python对象：

import pickle # 定义复杂对象 class User: def __init__(self, name, age): self.name = name self.age = age def greet(self): return f"Hello, {self.name}" # 序列化对象 user = User("John", 30) pickle_data = pickle.dumps(user) # 反序列化 restored_user = pickle.loads(pickle_data) print(restored_user.greet()) # 输出: Hello, John # 保存到文件 with open('user.pkl', 'wb') as f: pickle.dump(user, f) # 从文件加载 with open('user.pkl', 'rb') as f: loaded_user = pickle.load(f)

二、第三方序列化库

2.1 MessagePack

MessagePack是一种高效的二进制格式，比JSON更小更快：

import msgpack # 基本使用 data = {'name': 'John', 'age': 30, 'scores': [95, 87, 92]} # 序列化 packed = msgpack.packb(data) print(f"Size: {len(packed)} bytes") # 反序列化 unpacked = msgpack.unpackb(packed) print(unpacked['name']) # 输出: John # 处理datetime import datetime from msgpack import Packer, Unpacker packer = Packer(default=lambda obj: obj.isoformat() if isinstance(obj, datetime.datetime) else obj) packed = packer.pack({'time': datetime.datetime.now()})

2.2 Protocol Buffers

Protocol Buffers是Google开发的高效序列化格式：

# 首先定义.proto文件 # message User { # string name = 1; # int32 age = 2; # repeated int32 scores = 3; # } # 使用生成的代码 import user_pb2 user = user_pb2.User() user.name = "John" user.age = 30 user.scores.extend([95, 87, 92]) # 序列化 data = user.SerializeToString() print(f"Size: {len(data)} bytes") # 反序列化 new_user = user_pb2.User() new_user.ParseFromString(data) print(new_user.name) # 输出: John

2.3 Apache Avro

Avro提供schema定义和高效的二进制序列化：

import avro.schema from avro.datafile import DataFileReader, DataFileWriter from avro.io import DatumReader, DatumWriter # 定义schema schema = avro.schema.parse(""" { "type": "record", "name": "User", "fields": [ {"name": "name", "type": "string"}, {"name": "age", "type": "int"}, {"name": "scores", "type": {"type": "array", "items": "int"}} ] } """) # 序列化 writer = DataFileWriter(open("users.avro", "wb"), DatumWriter(), schema) writer.append({"name": "John", "age": 30, "scores": [95, 87, 92]}) writer.close() # 反序列化 reader = DataFileReader(open("users.avro", "rb"), DatumReader()) for user in reader: print(user) reader.close()

三、序列化方案对比

3.1 性能对比实验

import json import pickle import msgpack import time # 测试数据 test_data = { 'name': 'John Doe', 'age': 30, 'email': 'john@example.com', 'scores': list(range(100)), 'metadata': {'active': True, 'role': 'admin'} } # 测试序列化性能 def test_serialization(name, serialize_func, deserialize_func): start = time.time() for _ in range(10000): serialized = serialize_func(test_data) serialize_time = time.time() - start start = time.time() for _ in range(10000): deserialize_func(serialized) deserialize_time = time.time() - start print(f"{name}:") print(f" Serialize: {serialize_time:.3f}s") print(f" Deserialize: {deserialize_time:.3f}s") print(f" Size: {len(serialized)} bytes") test_serialization("JSON", json.dumps, json.loads) test_serialization("pickle", pickle.dumps, pickle.loads) test_serialization("msgpack", msgpack.packb, msgpack.unpackb)

3.2 方案选择指南

特性	JSON	pickle	MessagePack	Protocol Buffers
可读性	高	无	无	无
跨语言	是	否	是	是
速度	中等	快	很快	很快
大小	中等	中等	小	很小
安全性	高	低	高	高
对象支持	基本类型	几乎所有	基本类型	需定义schema

四、生产环境最佳实践

4.1 安全注意事项

# 危险：不要反序列化不受信任的数据 import pickle import io # 安全做法：使用限制的反序列化 class RestrictedUnpickler(pickle.Unpickler): def find_class(self, module, name): # 只允许特定类 if module == '__main__' and name == 'User': return User raise pickle.UnpicklingError(f"禁止加载 {module}.{name}") # 使用安全的反序列化 data = b'...' # 来自不受信任的来源 try: obj = RestrictedUnpickler(io.BytesIO(data)).load() except pickle.UnpicklingError as e: print(f"安全错误: {e}")

4.2 版本兼容性

# 处理数据格式版本 def serialize_with_version(data): return { 'version': 2, 'data': data } def deserialize_with_version(raw_data): parsed = json.loads(raw_data) version = parsed.get('version', 1) if version == 1: # 转换旧格式 return migrate_v1_to_v2(parsed['data']) elif version == 2: return parsed['data'] else: raise ValueError(f"不支持的版本: {version}")

4.3 大数据序列化

# 流式处理大数据 import json import msgpack def stream_serialize(data_generator, output_file): """流式序列化大量数据""" with open(output_file, 'wb') as f: packer = msgpack.Packer() for item in data_generator: f.write(packer.pack(item)) def stream_deserialize(input_file): """流式反序列化""" with open(input_file, 'rb') as f: unpacker = msgpack.Unpacker(f) for item in unpacker: yield item # 使用示例 def generate_large_data(): for i in range(1000000): yield {'id': i, 'value': f"data_{i}"} stream_serialize(generate_large_data(), 'large_data.msgpack') for item in stream_deserialize('large_data.msgpack'): process_item(item)

五、高级序列化技术

5.1 自定义序列化协议

class CustomSerializer: """自定义二进制序列化器""" @staticmethod def serialize(obj): if isinstance(obj, dict): return CustomSerializer._serialize_dict(obj) elif isinstance(obj, list): return CustomSerializer._serialize_list(obj) elif isinstance(obj, str): return CustomSerializer._serialize_str(obj) elif isinstance(obj, int): return CustomSerializer._serialize_int(obj) else: raise ValueError(f"不支持的类型: {type(obj)}") @staticmethod def _serialize_dict(d): result = bytearray([0x01]) # 字典标记 result += len(d).to_bytes(4, 'big') for k, v in d.items(): result += CustomSerializer.serialize(k) result += CustomSerializer.serialize(v) return bytes(result) # ... 其他序列化方法

5.2 压缩与序列化结合

import gzip import msgpack def compress_and_serialize(data): """压缩并序列化""" packed = msgpack.packb(data) compressed = gzip.compress(packed) return compressed def decompress_and_deserialize(compressed_data): """解压并反序列化""" packed = gzip.decompress(compressed_data) return msgpack.unpackb(packed) # 使用示例 original = {'data': list(range(1000))} compressed = compress_and_serialize(original) print(f"原始大小: {len(msgpack.packb(original))} bytes") print(f"压缩后大小: {len(compressed)} bytes") restored = decompress_and_deserialize(compressed)