在现代软件开发中，数据序列化是一个至关重要的技术环节，它负责将复杂的程序对象转换为可传输和存储的格式。JSON作为最广泛使用的数据交换格式，在Web服务、API接口和数据持久化中发挥着核心作用。然而，Python标准库中的JSON模块仅支持基本数据类型的序列化，面对复杂的自定义对象时往往力不从心。

基本原理与挑战

JSON序列化本质上是一个将内存中的对象表示转换为字符串格式的过程。Python的标准json模块基于递归下降的方式处理数据结构，它能够自动识别并序列化字典、列表、字符串、数字、布尔值和None等基本类型。这种机制的核心在于类型检测和格式转换，通过遍历对象的内部结构来生成对应的JSON表示。

当面对自定义类实例、日期时间对象、集合类型或其他复杂数据结构时，标准JSON模块会抛出TypeError异常。这是因为JSON规范本身只定义了有限的数据类型，无法直接表示Python中丰富的对象类型。解决这一挑战的关键在于建立对象到JSON表示的映射关系，将复杂对象的内部状态提取出来，转换为JSON支持的基本类型。

自定义编码器实现

实现自定义JSON编码器的核心方法是继承json.JSONEncoder类并重写其default方法。这个方法在遇到无法序列化的对象时被调用，可以提供自定义的序列化逻辑。

下面的实现展示了一个完整的自定义编码器，能够处理日期时间对象、集合类型、自定义类实例等多种复杂情况。

import json
import datetime
from decimal import Decimal
from dataclasses import dataclass

class CustomJSONEncoder(json.JSONEncoder):
    """
    自定义JSON编码器，支持多种复杂数据类型的序列化
    处理日期时间、集合、自定义对象等类型
    """
    
    def default(self, obj):
        # 处理日期时间对象
        if isinstance(obj, datetime.datetime):
            return {
                '__type__': 'datetime',
                'value': obj.isoformat()
            }
        
        if isinstance(obj, datetime.date):
            return {
                '__type__': 'date',
                'value': obj.isoformat()
            }
        
        # 处理集合类型
        if isinstance(obj, set):
            return {
                '__type__': 'set',
                'value': list(obj)
            }
        
        if isinstance(obj, tuple):
            return {
                '__type__': 'tuple',
                'value': list(obj)
            }
        
        # 处理Decimal类型
        if isinstance(obj, Decimal):
            return {
                '__type__': 'decimal',
                'value': str(obj)
            }
        
        # 处理自定义对象
        if hasattr(obj, '__dict__'):
            return {
                '__type__': 'custom_object',
                '__class__': obj.__class__.__name__,
                'attributes': obj.__dict__
            }
        
        # 处理dataclass对象
        if hasattr(obj, '__dataclass_fields__'):
            return {
                '__type__': 'dataclass',
                '__class__': obj.__class__.__name__,
                'fields': {field.name: getattr(obj, field.name) 
                          for field in obj.__dataclass_fields__.values()}
            }
        
        return super().default(obj)

# 定义测试类
@dataclass
class Person:
    name: str
    age: int
    email: str

class Product:
    def __init__(self, name, price, tags):
        self.name = name
        self.price = price
        self.tags = tags
        self.created_at = datetime.datetime.now()

# 创建测试数据
test_data = {
    'person': Person('张三', 30, 'zhangsan@example.com'),
    'product': Product('智能手机', Decimal('2999.99'), {'电子产品', '通讯设备'}),
    'timestamp': datetime.datetime.now(),
    'numbers': (1, 2, 3, 4, 5)
}

# 使用自定义编码器进行序列化
json_string = json.dumps(test_data, cls=CustomJSONEncoder, indent=2, ensure_ascii=False)
print("序列化结果:")
print(json_string)

运行结果：

序列化结果:
{
  "person": {
    "__type__": "custom_object",
    "__class__": "Person",
    "attributes": {
      "name": "张三",
      "age": 30,
      "email": "zhangsan@example.com"
    }
  },
  "product": {
    "__type__": "custom_object",
    "__class__": "Product",
    "attributes": {
      "name": "智能手机",
      "price": {
        "__type__": "decimal",
        "value": "2999.99"
      },
      "tags": {
        "__type__": "set",
        "value": [
          "电子产品",
          "通讯设备"
        ]
      },
      "created_at": {
        "__type__": "datetime",
        "value": "2025-06-08T12:59:16.355264"
      }
    }
  },
  "timestamp": {
    "__type__": "datetime",
    "value": "2025-06-08T12:59:16.355270"
  },
  "numbers": [
    1,
    2,
    3,
    4,
    5
  ]
}

高级编码器

为了构建更加强大的序列化系统，需要实现循环引用检测、深度限制和选择性序列化等高级功能。

下面的实现展示了一个功能完整的高级编码器，提供了生产环境所需的各种特性。

import datetime
import json


class AdvancedJSONEncoder(json.JSONEncoder):
    """
    高级JSON编码器，支持循环引用检测、深度限制等功能
    """

    def __init__(self, *args, **kwargs):
        self.max_depth = kwargs.pop('max_depth', 10)
        self.skip_private = kwargs.pop('skip_private', True)
        super().__init__(*args, **kwargs)
        self._obj_tracker = set()
        self._current_depth = 0

    def encode(self, obj):
        self._obj_tracker.clear()
        self._current_depth = 0
        return super().encode(obj)

    def default(self, obj):
        # 深度检查
        if self._current_depth > self.max_depth:
            return f"<深度超限>"

        # 循环引用检查
        obj_id = id(obj)
        if obj_id in self._obj_tracker:
            return f"<循环引用: {type(obj).__name__}>"

        self._obj_tracker.add(obj_id)
        self._current_depth += 1

        try:
            # 处理日期时间
            if isinstance(obj, datetime.datetime):
                return {'__type__': 'datetime', 'value': obj.isoformat()}

            # 处理自定义对象
            if hasattr(obj, '__dict__'):
                attributes = {}
                for key, value in obj.__dict__.items():
                    if self.skip_private and key.startswith('_'):
                        continue
                    if not callable(value):
                        attributes[key] = value
                return {
                    '__type__': 'custom_object',
                    '__class__': obj.__class__.__name__,
                    'attributes': attributes
                }

            return str(obj)

        finally:
            self._obj_tracker.discard(obj_id)
            self._current_depth -= 1


# 测试高级编码器
class Person:
    def __init__(self, name, age, email):
        self.name = name
        self.age = age
        self.email = email
        self._internal_id = "person_001"


class Department:
    def __init__(self, name):
        self.name = name
        self.employees = []
        self._internal_id = "dept_001"

    def add_employee(self, employee):
        self.employees.append(employee)


dept = Department("技术部")
person = Person("李四", 25, "lisi@example.com")
dept.add_employee(person)

encoder = AdvancedJSONEncoder(indent=2, ensure_ascii=False, max_depth=5, skip_private=True)
result = encoder.encode(dept)
print("高级编码器结果:")
print(result)

运行结果：

高级编码器结果:
{
  "__type__": "custom_object",
  "__class__": "Department",
  "attributes": {
    "name": "技术部",
    "employees": [
      {
        "__type__": "custom_object",
        "__class__": "Person",
        "attributes": {
          "name": "李四",
          "age": 25,
          "email": "lisi@example.com"
        }
      }
    ]
  }
}

反序列化机制实现

完整的序列化解决方案还需要支持从JSON到对象的反向转换。通过实现自定义的object_hook函数，可以在JSON解析过程中识别特殊的类型标记，并执行相应的对象重构逻辑。

import datetime
import json
from decimal import Decimal


class CustomJSONEncoder(json.JSONEncoder):
    """
    自定义JSON编码器，支持多种复杂数据类型的序列化
    处理日期时间、集合、自定义对象等类型
    """

    def default(self, obj):
        # 处理日期时间对象
        if isinstance(obj, datetime.datetime):
            return {
                '__type__': 'datetime',
                'value': obj.isoformat()
            }

        if isinstance(obj, datetime.date):
            return {
                '__type__': 'date',
                'value': obj.isoformat()
            }

        # 处理集合类型
        if isinstance(obj, set):
            return {
                '__type__': 'set',
                'value': list(obj)
            }

        if isinstance(obj, tuple):
            return {
                '__type__': 'tuple',
                'value': list(obj)
            }

        # 处理Decimal类型
        if isinstance(obj, Decimal):
            return {
                '__type__': 'decimal',
                'value': str(obj)
            }

        # 处理自定义对象
        if hasattr(obj, '__dict__'):
            return {
                '__type__': 'custom_object',
                '__class__': obj.__class__.__name__,
                'attributes': obj.__dict__
            }

        # 处理dataclass对象
        if hasattr(obj, '__dataclass_fields__'):
            return {
                '__type__': 'dataclass',
                '__class__': obj.__class__.__name__,
                'fields': {field.name: getattr(obj, field.name)
                           for field in obj.__dataclass_fields__.values()}
            }

        return super().default(obj)

class JSONDecoder:
    """
    自定义JSON解码器，支持对象反序列化
    """

    def __init__(self):
        self.type_handlers = {
            'datetime': self._decode_datetime,
            'date': self._decode_date,
            'set': self._decode_set,
            'tuple': self._decode_tuple,
            'decimal': self._decode_decimal
        }

    def decode(self, json_string):
        return json.loads(json_string, object_hook=self._object_hook)

    def _object_hook(self, obj):
        if '__type__' in obj:
            type_name = obj['__type__']
            if type_name in self.type_handlers:
                return self.type_handlers[type_name](obj)
        return obj

    def _decode_datetime(self, obj):
        return datetime.datetime.fromisoformat(obj['value'])

    def _decode_date(self, obj):
        return datetime.date.fromisoformat(obj['value'])

    def _decode_set(self, obj):
        return set(obj['value'])

    def _decode_tuple(self, obj):
        return tuple(obj['value'])

    def _decode_decimal(self, obj):
        return Decimal(obj['value'])


# 测试完整的序列化和反序列化
original_data = {
    'timestamp': datetime.datetime.now(),
    'price': Decimal('99.99'),
    'tags': {'python', 'json', 'serialization'},
    'coordinates': (10, 20, 30)
}

# 序列化
json_data = json.dumps(original_data, cls=CustomJSONEncoder)
print("序列化:", json_data)

# 反序列化
decoder = JSONDecoder()
restored_data = decoder.decode(json_data)
print("反序列化成功，时间类型:", type(restored_data['timestamp']))

运行结果：

序列化: {"timestamp": {"__type__": "datetime", "value": "2025-06-08T13:03:42.075846"}, "price": {"__type__": "decimal", "value": "99.99"}, "tags": {"__type__": "set", "value": ["json", "serialization", "python"]}, "coordinates": [10, 20, 30]}
反序列化成功，时间类型: <class 'datetime.datetime'>

总结

自定义JSON编码器为Python应用程序提供了强大的数据序列化能力。通过扩展标准库的功能，我们能够处理复杂的对象结构，实现完整的数据持久化和传输方案。在实际应用中，需要注意安全性考虑，建立白名单机制来限制可重建的类型。同时要考虑性能优化，避免过度复杂的序列化逻辑影响系统效率。合理使用自定义JSON编码器，能够显著提升系统的数据处理能力，为构建可扩展的现代应用奠定坚实基础。通过掌握这些技术，开发者可以更好地应对复杂的数据序列化需求，构建高质量的Python应用程序。