第三十九篇：序列化——pickle 与 JSON
前言
**序列化（Serialization）**是将 Python 对象转换为可存储或传输的格式，反序列化则是将其还原为原始对象。Python 提供了多种序列化方案，其中最常用的是 pickle（原生 Python 对象）和 JSON（跨语言通用格式）。本文将系统讲解这两大序列化工具。
pickle —— Python 对象序列化
基本用法
import pickle
# 序列化：Python 对象 -> bytes
data = {
"name": "小明",
"age": 25,
"scores": [95, 88, 92, 76, 100],
"active": True,
}
pickled = pickle.dumps(data) # 转为 bytes
print(f"原始大小: {len(str(data))} 字符")
print(f"pickle 大小: {len(pickled)} 字节")
# pickle 大小: 74 字节
# 反序列化：bytes -> Python 对象
loaded = pickle.loads(pickled)
print(loaded)
# {'name': '小明', 'age': 25, 'scores': [95, 88, 92, 76, 100], 'active': True}
print(type(loaded)) # <class 'dict'>
print(loaded is data) # False ← 是新对象
文件持久化
import pickle
def save_object(obj, filepath):
"""保存任意 Python 对象到文件"""
with open(filepath, "wb") as f:
pickle.dump(obj, f, protocol=pickle.HIGHEST_PROTOCOL)
print(f"已保存: {filepath}")
def load_object(filepath):
"""从文件加载 Python 对象"""
with open(filepath, "rb") as f:
return pickle.load(f)
# 保存复杂数据
complex_data = {
"users": [
{"name": "小明", "score": 95, "tags": ["优秀", "数学"]},
{"name": "小红", "score": 88, "tags": ["进步中", "英语"]},
],
"metadata": {"version": "1.0", "created": "2026-09-09"},
}
save_object(complex_data, "data.pkl")
loaded = load_object("data.pkl")
print(loaded["users"][0]["tags"]) # ['优秀', '数学']
pickle 协议版本
import pickle
data = {"hello": "world"}
for protocol in range(pickle.HIGHEST_PROTOCOL + 1):
pickled = pickle.dumps(data, protocol=protocol)
print(f"协议 v{protocol}: {len(pickled)} 字节")
# 协议 v0: 27 字节 (ASCII 文本，最慢)
# 协议 v1: 二进制格式
# 协议 v2: Python 2.3+（推荐最低版本）
# 协议 v3: Python 3.0+（默认）
# 协议 v4: Python 3.4+（支持超大对象）
# 协议 v5: Python 3.8+（最新，最大）
使用 HIGHEST_PROTOCOL 获得最佳性能
import pickle
import time
large_list = list(range(1_000_000))
# 不同协议的对比
protocols = [0, 2, 3, 4, 5]
for protocol in protocols:
start = time.perf_counter()
pickled = pickle.dumps(large_list, protocol=protocol)
loaded = pickle.loads(pickled)
elapsed = time.perf_counter() - start
print(f"v{protocol}: {len(pickled)//1024:6d} KB | {elapsed*1000:.1f}ms")
# v0: 7865 KB | 850.2ms
# v2: 763 KB | 52.3ms
# v3: 763 KB | 48.7ms
# v4: 763 KB | 35.1ms
# v5: 763 KB | 32.8ms
序列化自定义类
import pickle
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
def distance_to(self, other):
import math
return math.sqrt((self.x - other.x)**2 + (self.y - other.y)**2)
def __repr__(self):
return f"Point({self.x}, {self.y})"
p1 = Point(3, 4)
p2 = Point(0, 0)
print(p1.distance_to(p2)) # 5.0
# 自定义序列化行为
class SerializablePoint(Point):
def to_dict(self):
return {"x": self.x, "y": self.y}
@classmethod
def from_dict(cls, data):
return cls(data["x"], data["y"])
sp = SerializablePoint(3, 4)
pickled = pickle.dumps(sp)
loaded = pickle.loads(pickled)
print(loaded.distance_to(Point(0, 0))) # 5.0
处理循环引用和复杂结构
import pickle
# 链表节点（有循环引用）
class Node:
def __init__(self, value):
self.value = value
self.next = None
self.prev = None
a = Node("A")
b = Node("B")
c = Node("C")
a.next = b
b.prev = a
b.next = c
c.prev = b
# pickle 自动处理循环引用
data = {"head": a, "length": 3}
pickled = pickle.dumps(data)
loaded = pickle.loads(pickled)
# 遍历验证
current = loaded["head"]
for i in range(3):
print(current.value, end=" ")
current = current.next
# A B C
# 验证双向链接
print("\n反向:")
current = loaded["head"].next.next # C
for i in range(3):
print(current.value, end=" ")
current = current.prev
# C B A
pickle 的安全注意事项
import pickle
# ❌ 永远不要 unpickle 不可信的数据！
# pickle 可以执行任意代码，存在严重安全风险
# pickle.loads(b'\x80\x04\x95...\x93...') ← 可能包含恶意代码
# ✅ 安全措施：
# 1. 只对可信数据使用 pickle
# 2. 使用 hash 校验文件完整性
import hashlib
def safe_save(obj, filepath):
data = pickle.dumps(obj, protocol=5)
checksum = hashlib.sha256(data).hexdigest()
with open(filepath, "wb") as f:
f.write(checksum.encode())
f.write(data)
def safe_load(filepath):
with open(filepath, "rb") as f:
saved_checksum = f.read(64).decode()
data = f.read()
actual_checksum = hashlib.sha256(data).hexdigest()
if saved_checksum != actual_checksum:
raise ValueError("文件已被篡改，校验失败!")
return pickle.loads(data)
JSON —— 跨语言通用格式
基本用法
import json
data = {
"name": "小明",
"age": 25,
"scores": [95, 88, 92],
"active": True,
"address": None,
}
# 序列化：Python 对象 -> JSON 字符串
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)
#
#
# {
# "name": "\u5c0f\u660e",
# ...
# }
# 注意：默认情况下 ensure_ascii=True，中文会变成 Unicode 转义
# 设置 ensure_ascii=False 保留原始中文
# 反序列化：JSON 字符串 -> Python 对象
loaded = json.loads(json_str)
print(loaded["name"]) # 小明
JSON 与 Python 类型映射
| Python 类型 | JSON 类型 |
|---|---|
dict |
object |
list / tuple |
array |
str |
string |
int, float |
number |
True / False |
true / false |
None |
null |
import json
mapping = [
({}, "对象 {}"),
([], "数组 []"),
("hello", '字符串 "hello"'),
(42, "数字 42"),
(3.14, "数字 3.14"),
(True, "布尔值 true"),
(False, "布尔值 false"),
(None, "空值 null"),
((1, 2), "元组转为数组 [1, 2]"),
]
for py_obj, expected in mapping:
result = json.dumps(py_obj)
print(f"{type(py_obj).__name__:8s} → {result:15s} （{expected}）")
文件读写 JSON
import json
config = {
"app_name": "MyApp",
"version": "2.0.0",
"debug": True,
"database": {
"host": "localhost",
"port": 5432,
"name": "production",
},
"features": ["auth", "logging", "cache", "monitoring"],
"limits": {
"max_connections": 100,
"timeout": 30,
},
}
# 写入 JSON 文件
with open("config.json", "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
print("配置文件已保存")
# 读取 JSON 文件
with open("config.json", "r", encoding="utf-8") as f:
loaded_config = json.load(f)
print(f"应用: {loaded_config['app_name']} v{loaded_config['version']}")
print(f"特性: {', '.join(loaded_config['features'])}")
处理特殊数据类型
import json
# ❌ JSON 不支持的类型
# set、datetime、bytes、自定义类等无法直接序列化
from datetime import datetime
from decimal import Decimal
data = {
"timestamp": datetime.now(),
"amount": Decimal("1234.56"),
"tags": {"python", "json"}, # set 会被转换为 dict keys
}
# 解决方法一：手动转换
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return {"__type__": "datetime", "__value__": obj.isoformat()}
if isinstance(obj, Decimal):
return {"__type__": "decimal", "__value__": str(obj)}
if isinstance(obj, set):
return {"__type__": "set", "__value__": list(obj)}
return super().default(obj)
json_str = json.dumps(data, cls=CustomEncoder, ensure_ascii=False, indent=2)
print(json_str)
#
#
# {
# "timestamp": {
# "__type__": "datetime",
# "__value__": "2026-09-09T12:00:00.000000"
# },
# ...
# }
# 解决方法二：简单的手动转换函数
def serialize_with_helpers(obj):
if isinstance(obj, datetime):
return f"@DATETIME:{obj.isoformat()}"
if isinstance(obj, Decimal):
return f"@DECIMAL:{str(obj)}"
if isinstance(obj, set):
return f"@SET:{','.join(sorted(obj))}"
raise TypeError(f"无法序列化 {type(obj)}")
def deserialize_with_helpers(obj):
if isinstance(obj, str) and obj.startswith("@DATETIME:"):
return datetime.fromisoformat(obj.replace("@DATETIME:", ""))
if isinstance(obj, str) and obj.startswith("@DECIMAL:"):
return Decimal(obj.replace("@DECIMAL:", ""))
if isinstance(obj, str) and obj.startswith("@SET:"):
return set(obj.replace("@SET:", "").split(","))
if isinstance(obj, dict) and "__type__" in obj:
if obj["__type__"] == "datetime":
return datetime.fromisoformat(obj["__value__"])
if obj["__type__"] == "decimal":
return Decimal(obj["__value__"])
if obj["__type__"] == "set":
return set(obj["__value__"])
return obj
格式化选项详解
import json
data = {"users": [{"name": "张三", "age": 25}, {"name": "李四", "age": 30}]}
# sort_keys: 按键排序
print(json.dumps(data, sort_keys=True, indent=2))
# separators: 自定义分隔符（紧凑输出）
compact = json.dumps(data, separators=(",", ":"))
print(compact) # {"users":[{"name":"张三","age":25},{"name":"李四","age":30}]}
# 压缩 JSON（用于 API 传输）
compressed = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
print(f"压缩后: {len(compressed)} 字节")
# 美化 JSON（用于展示）
pretty = json.dumps(data, indent=2, ensure_ascii=False, sort_keys=True)
print(pretty)
简易查询
import json
records = [
{"id": 1, "name": "键盘", "price": 299, "category": "外设"},
{"id": 2, "name": "鼠标", "price": 99, "category": "外设"},
{"id": 3, "name": "显示器", "price": 1999, "category": "显示设备"},
{"id": 4, "name": "耳机", "price": 199, "category": "音频"},
{"id": 5, "name": "摄像头", "price": 399, "category": "音视频"},
]
# 筛选单价低于 300 的产品
cheap_products = [item for item in records if item["price"] < 300]
print([p["name"] for p in cheap_products])
# ['键盘', '鼠标', '耳机']
# 按类别分组
from collections import defaultdict
categories = defaultdict(list)
for item in records:
categories[item["category"]].append(item["name"])
for cat, items in categories.items():
print(f"{cat}: {', '.join(items)}")
# 外设: 键盘, 鼠标
# 显示设备: 显示器
# 音频: 耳机
# 音视频: 摄像头
pickle vs JSON 对比
| 特性 | pickle | JSON |
|---|---|---|
| 支持类型 | 所有 Python 对象 | 基础类型 + 部分扩展 |
| 跨语言 | ❌ 仅 Python | ✅ 几乎所有语言 |
| 可读性 | ❌ 二进制 | ✅ 人类可读（文本） |
| 安全性 | ⚠️ 不安全（可执行代码） | ✅ 安全（纯数据） |
| 速度 | 快 | 中等 |
| 文件大小 | 较小 | 较大 |
| 适用场景 | 本地缓存、模型保存 | API 交互、配置文件 |
实战：数据库查询结果缓存
import json
import pickle
import hashlib
import time
from pathlib import Path
class HybridCache:
"""混合缓存：JSON（配置/索引）+ pickle（计算结果）"""
def __init__(self, cache_dir="cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.meta_file = self.cache_dir / "index.json"
self._load_meta()
def _load_meta(self):
"""加载索引"""
if self.meta_file.exists():
with open(self.meta_file, "r", encoding="utf-8") as f:
self.meta = json.load(f)
else:
self.meta = {"entries": {}, "stats": {"hits": 0, "misses": 0}}
def _save_meta(self):
"""保存索引"""
with open(self.meta_file, "w", encoding="utf-8") as f:
json.dump(self.meta, f, ensure_ascii=False, indent=2)
def _make_key(self, query):
return hashlib.sha256(query.encode("utf-8")).hexdigest()
def get(self, query, loader_func=None):
"""获取缓存或加载新数据"""
key = self._make_key(query)
# 检查是否在索引中
entry = self.meta["entries"].get(key)
if entry:
pickle_file = self.cache_dir / f"{key}.pkl"
if pickle_file.exists():
with open(pickle_file, "rb") as f:
cached_time, result = pickle.load(f)
# 可选：设置 TTL
ttl = entry.get("ttl", 3600)
if time.time() - cached_time > ttl:
self.delete(query)
self.meta["stats"]["misses"] += 1
self._save_meta()
return None
self.meta["stats"]["hits"] += 1
self._save_meta()
print(f"[缓存命中] {query[:50]}...")
return result
# 缓存未命中，调用加载函数
if loader_func:
result = loader_func(query)
self.set(query, result)
else:
self.meta["stats"]["misses"] += 1
self._save_meta()
return None
return self.get(query)
def set(self, query, result, ttl=3600):
"""存入缓存"""
key = self._make_key(query)
pickle_file = self.cache_dir / f"{key}.pkl"
with open(pickle_file, "wb") as f:
pickle.dump((time.time(), result), f)
self.meta["entries"][key] = {
"created_at": time.time(),
"ttl": ttl,
}
self._save_meta()
def delete(self, query):
"""删除缓存条目"""
key = self._make_key(query)
self.meta["entries"].pop(key, None)
pickle_file = self.cache_dir / f"{key}.pkl"
if pickle_file.exists():
pickle_file.unlink()
self._save_meta()
def stats(self):
total = self.meta["stats"]["hits"] + self.meta["stats"]["misses"]
hit_rate = self.meta["stats"]["hits"] / total * 100 if total > 0 else 0
return {
"entries": len(self.meta["entries"]),
"total_requests": total,
"hit_rate": f"{hit_rate:.1f}%",
}
# 模拟数据库查询（耗时的操作）
def expensive_query(sql):
print(f"[数据库查询] {sql[:50]}...")
time.sleep(0.5)
return [{"id": 1, "name": "结果", "sql": sql}]
cache = HybridCache("product_cache")
sql = "SELECT * FROM products WHERE category = 'electronics' AND price < 500"
print("第一次查询:")
result = cache.get(sql, loader_func=lambda q: expensive_query(q))
print(cache.stats())
print("\n第二次查询（应该命中缓存）:")
result = cache.get(sql)
print(cache.stats())
实战：CSV 替代方案——结构化数据存储
import json
def export_to_jsonl(records, filepath):
"""导出为 JSON Lines 格式（每行一个 JSON 对象）"""
with open(filepath, "w", encoding="utf-8") as f:
for record in records:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
print(f"已导出 {len(records)} 条记录到 {filepath}")
def import_from_jsonl(filepath):
"""导入 JSON Lines 格式"""
records = []
with open(filepath, "r", encoding="utf-8") as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if line:
try:
records.append(json.loads(line))
except json.JSONDecodeError as e:
print(f"警告: 第 {line_num} 行解析失败: {e}")
return records
# 生成测试数据
products = [
{"id": i, "name": f"产品{i}", "price": round(100 + i * 50, 2),
"tags": ["热销", "新品"][i % 2]}
for i in range(1, 21)
]
export_to_jsonl(products, "products.jsonl")
loaded = import_from_jsonl("products.jsonl")
print(f"共加载 {len(loaded)} 条记录")
# 统计
prices = [p["price"] for p in loaded]
print(f"平均价格: ¥{sum(prices)/len(prices):.2f}")
print(f"最高价: ¥{max(prices):.2f}")
最佳实践总结
✅ 推荐做法
- 对外接口用 JSON — API、配置文件、跨平台数据交换
- 内部缓存用 pickle — 加速频繁读写，保持 Python 类型保真度
- 大数据用 JSON Lines — 流式读写，适合日志和批量数据
- 敏感数据加密存储 — pickle 本身不解密，需要自行加盐加密
- JSON 指定编码 — 始终使用
encoding="utf-8"
❌ 避免做法
import pickle
import json
# ❌ 对不可信来源 unpickle
user_input_pickle = input("输入 pickle 数据: ")
data = pickle.loads(user_input_pickle) # 可能被注入恶意代码!
# ❌ 忽略 JSON 解码错误
try:
data = json.loads(invalid_json_string)
except json.JSONDecodeError:
pass # 默默失败可能导致后续逻辑出错
# ✅ 正确做法：明确处理错误
try:
data = json.loads(invalid_json_string)
except json.JSONDecodeError as e:
print(f"JSON 解析失败: {e.msg} (第 {e.lineno} 行)")
data = get_default_value()
# ❌ 忘记 ensure_ascii=False
json.dumps({"中文名": "测试"})
# {"\\u4e2d\\u6587\\u540d": "\\u6d4b\\u8bd5"} ← 中文被转义
# ✅ 设置 ensure_ascii=False
json.dumps({"中文名": "测试"}, ensure_ascii=False)
# {"中文名": "测试"}
总结
本文系统讲解了 Python 序列化的两大核心工具：
- pickle — Python 原生对象序列化，支持所有类型，速度快，但仅限 Python
- JSON — 跨语言通用格式，人类可读，安全，适合 API 和配置
- 协议选择 — HIGHEST_PROTOCOL 最优性能，ensure_ascii=False 保留中文
- 自定义序列化 — JSONEncoder/Decoder 处理特殊类型
- 安全考量 — pickle 不信任数据源，JSON 处理解码异常
- 混合缓存方案 — JSON 存索引 + pickle 存数据的实用架构
- JSON Lines — 流式格式的批量数据处理
下一篇文章将深入讲解正则表达式：从入门到实战。
本文是 Python 教程系列的第三十九篇。