Python中序列化操作

一、序列化

概念：将内存中的数据结构或对象转换为可存储/传输格式（字节流/字符串）的过程。
反序列化：将序列化数据还原为原始对象的过程。
核心作用：
- 持久化存储：保存程序状态（如机器学习模型、用户配置）
- 跨系统通信：网络传输（API交互、微服务）
- 跨语言交换：JSON/MessagePack等格式实现多语言兼容

二、JSON（JavaScript Object Notation）

Python与JSON类型映射

Python类型	JSON等效类型	注意事项
`dict`	`{}`	键必须是字符串
`list`/`tuple`	`[]`	元组反序列化后变为列表
`str`	`"string"`	非ASCII字符需设`ensure_ascii=False`
`int`/`float`	`123.45`	JSON数字有精度限制
`True`/`False`	`true`/`false`	大小写敏感
`None`	`null`

除了set类型以及自定义类型（可以通过自定义JSONEncoder的方式进行解析）外，都可以直接进行序列化操作，否则会出现TypeError

对于大整型数据建议使用字符串形式传输，避免精度丢失

三、主流序列化库对比

库名	数据格式	速度	体积	数据类型支持	跨语言	安全风险
`JSON`	文本	中等	中等	基础类型	✓	低
`Pickle`	二进制	快	大	所有Python对象	✗	高
`MessagePack`	二进制	极快	小	基础类型+部分扩展	✓	中
`ujson`	文本	快	中等	基础类型	✓	中
`orjson`	二进制	极快	小	基础类型+datetime	✓	低
`simdjson`	文本	反序列化极快	中等	基础类型	✓	低

四、各自使用示例

1. JSON（标准库）

特点：跨语言、可读性强，适合Web场景。

局限：不支持自定义类、函数、循环引用、二进制数据。

import json

# 序列化（Python对象 → JSON字符串）
data = {"name": "李雷", "age": 25, "skills": ["Python", "机器学习"]}
json_str = json.dumps(data, ensure_ascii=False, indent=4)  # 禁用ASCII编码保留中文
print("JSON字符串:", json_str)

# 写入文件
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, indent=4)

# 反序列化（JSON字符串 → Python对象）
loaded_data = json.loads(json_str)
print("解析后的年龄:", loaded_data["age"])

# 从文件读取
with open("data.json", "r", encoding="utf-8") as f:
    file_data = json.load(f)
print("文件读取的技能:", file_data["skills"])

或者自定义解析类：

# 自定义序列化（处理复杂对象）
class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        elif isinstance(obj, set):
            return list(obj)
        return super().default(obj)

json.dumps(data, cls=CustomEncoder)

# 循环引用处理
json.dumps(data, skipkeys=True)  # 跳过无法序列化的键

场景：RESTful API、配置文件。

关键参数：

ensure_ascii=False：允许非ASCII字符（如中文）正常显示而非\u编码。
indent：美化输出，增加缩进。
sort_keys=True：按键名排序输出

2. Pickle（Python专属）

特点：支持所有Python对象（类、函数等）。

风险：反序列化恶意数据可执行任意代码！

import pickle
# 自定义类
class User:
    def __init__(self, name, level):
        self.name = name
        self.level = level

# 序列化（含自定义对象）
user = User("韩梅梅", 99)
with open("user.pkl", "wb") as f:  # 必须二进制模式
    pickle.dump(user, f)

# 反序列化
with open("user.pkl", "rb") as f:
    loaded_user = pickle.load(f)
print("还原用户:", loaded_user.name, loaded_user.level)  # 输出：韩梅梅 99

场景：Python内部缓存、科学计算中间存储。

3. MessagePack（高效二进制）

特点：类似JSON结构，但速度更快、体积更小。

import msgpack

# 安装：pip install msgpack
data = {"id": 1, "tags": ("重要", "紧急"), "values": [1.2, 3.4]}
packed = msgpack.packb(data)  # 序列化为二进制
with open("msgpack.pack", "wb") as f:
  f.write(packed)
import json
json_str = json.dumps(data)
print("JSON长度:", len(json_str))
print("MessagePack长度:", len(packed))  # 比JSON小50%+

# 反序列化（元组→列表）
unpacked = msgpack.unpackb(packed)
print("还原标签类型:", type(unpacked["tags"]))  # 输出：<class 'list'>

# 增量解析（网络流场景）
unpacker = msgpack.Unpacker()
unpacker.feed(packed[:5])  # 模拟分块传输
unpacker.feed(packed[5:])
for obj in unpacker:
    print("流式解析:", obj["id"])

场景：物联网设备通信、高并发微服务。

4. 高性能JSON库

ujson

import ujson as json  # 替代标准库
json.dumps(data)  # 速度提升4.5倍

适用：高频API服务。

orjson

import orjson
orjson.dumps({"time": datetime.now()})  # 原生支持datetime

适用：含日期/数值的科学计算。

simdjson

import simdjson
parser = simdjson.Parser()
data = parser.parse(large_file.read())  # 1GB文件解析快10倍

适用：百万级JSON文件处理。

五、场景化选型建议

需求场景	推荐库	理由
前后端`API`交互	`JSON/orjson`	跨语言、易读、性能均衡
`Python`内部对象持久化	`Pickle`	支持所有类型，无转换成本
高并发网络传输（如微服务）	`MessagePack`	二进制高效，体积比`JSON`小50%+
日志/大文件处理	`simdjson`	反序列化极致优化，内存效率高
含日期/数值的科学数据	`orjson`	原生支持`datetime/numpy`

⚠️ 避坑指南：

安全性：Pickle仅反序列化可信数据！

兼容性：JSON避免使用元组（转为列表）

性能陷阱：大文件用simdjson分块解析，避免内存溢出

六、总结

序列化是系统开发的核心基础设施，选型需平衡四大维度：

数据兼容：Pickle > MessagePack > JSON
性能需求：MessagePack/orjson > ujson > JSON
安全要求：JSON > orjson > Pickle
跨平台性：JSON > MessagePack > Pickle

终极建议：

通用场景：标准库JSON（易用性强）

封闭系统：Pickle（功能全面）

性能瓶颈：MessagePack/orjson（速度极致优化）

数据科学：orjson+simdjson（数值/日期友好）

七、自定义序列化器

当然，自己也可以实现一个简易的序列化器方法，下面是Demo：

def simple_json_serialize(obj):
    # 处理None → null
    if obj is None:
        return "null"

    # 处理基础类型（int, float, bool）
    elif isinstance(obj, (int, float)):
        return str(obj)
    elif isinstance(obj, bool):
        return "true" if obj else "false"

    # 处理字符串（转义特殊字符）
    elif isinstance(obj, str):
        escaped = obj.replace('"', r'\"').replace('\\', r'\\')
        return f'"{escaped}"'

    # 处理列表 → [elem1, elem2]
    elif isinstance(obj, list):
        elements = [simple_json_serialize(e) for e in obj]
        return f'[{", ".join(elements)}]'

    # 处理字典 → { "k1": v1, "k2": v2 }
    elif isinstance(obj, dict):
        pairs = []
        for k, v in obj.items():
            key_str = f'"{k}"'  # 键必须是字符串
            val_str = simple_json_serialize(v)
            pairs.append(f"{key_str}: {val_str}")
        return f'{{ {", ".join(pairs)} }}'

    # 处理自定义对象（需实现to_dict方法）
    elif hasattr(obj, 'to_dict'):
        return simple_json_serialize(obj.to_dict())

    # 其他类型不支持，抛出异常
    else:
        raise TypeError(f"Unsupported type: {type(obj)}")

# 示例：自定义类
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

    def to_dict(self):
        return {"name": self.name, "age": self.age}

def serialize_via_reflection(obj):
    if hasattr(obj, "__dict__"):
        data = {}
        for attr_name, attr_value in obj.__dict__.items():
            # 过滤私有属性（以_开头）
            if not attr_name.startswith("_"):
                data[attr_name] = serialize_via_reflection(attr_value)
        return data
    elif isinstance(obj, (list, dict)):
        # 递归处理容器
        return [serialize_via_reflection(e) for e in obj] if isinstance(obj, list) \
               else {k: serialize_via_reflection(v) for k, v in obj.items()}
    else:
        return obj

# 测试
class User:
  def __init__(self, **args):
      self.name = args.get("name", "")
      self.age = args.get("age", 0)


data = {
    "name": "Alice",
    "age": 30,
    "scores": [95, 88, 92],
    "address": {"city": "Beijing", "zip": 100000},
    "friend": Person("Bob", 28)
}

json_str = simple_json_serialize(data)
# print(json_str)
# 输出：
# { 
#    "name": "Alice", 
#    "age": 30, 
#    "scores": [95, 88, 92], 
#    "address": { "city": "Beijing", "zip": 100000 }, 
#    "friend": { "name": "Bob", "age": 28 } 
# } 
# 结合之前的序列化函数
json_str = simple_json_serialize(serialize_via_reflection(User(**{"name":"AAA", "age":15})))
print(json_str)
with open("build_json.json", "w") as f:
  f.write(json_str)

import json
with open("build_json.json", "rb") as f:
  print(json.loads(f.read()))

结果：

{ "name": "AAA", "age": 15 }
{'name': 'AAA', 'age': 15}

代码下载地址：点击下载 json_demo.7z

K-ON

JSON序列化

Python中序列化操作

一、序列化

二、JSON（JavaScript Object Notation）

Python与JSON类型映射

三、主流序列化库对比

四、各自使用示例

1. JSON（标准库）

2. Pickle（Python专属）

3. MessagePack（高效二进制）

4. 高性能JSON库

五、场景化选型建议

六、总结

七、自定义序列化器

Administrator

Python中序列化操作

一、序列化

二、JSON（JavaScript Object Notation）

Python与JSON类型映射

三、主流序列化库对比

四、各自使用示例

1. JSON（标准库）

2. Pickle（Python专属）

3. MessagePack（高效二进制）

4. 高性能JSON库

五、场景化选型建议

六、总结

七、自定义序列化器

2025-8-2_周六

培训Day17

Administrator