从基础到高级详解Python对象序列化的实战指南

def demonstrate_serialization_concepts():
    """
    演示序列化的核心概念和用途
    """
    concepts = {
        '序列化 (Serialization)': '将对象转换为字节流或文本格式的过程',
        '反序列化 (Deserialization)': '从序列化格式重建原始对象的过程',
        '持久化 (Persistence)': '将对象状态保存到持久存储（如文件、数据库）',
        '数据交换 (Data Exchange)': '在不同系统或进程间传输对象数据',
        '深度复制 (Deep Copy)': '通过序列化/反序列化实现对象的深度复制',
        '状态恢复 (State Restoration)': '保存和恢复应用程序状态'
    }
    
    print("=== 序列化核心概念 ===")
    for term, definition in concepts.items():
        print(f"{term:20}: {definition}")
    
    # 序列化的常见格式
    formats = [
        ('二进制格式', '紧凑高效，但不易读', 'pickle, protobuf'),
        ('文本格式', '可读性好，但体积较大', 'JSON, XML, YAML'),
        ('混合格式', '平衡可读性和效率', 'MessagePack, BSON'),
        ('专用格式', '针对特定场景优化', 'Avro, Thrift')
    ]
    
    print("\n=== 常见序列化格式 ===")
    for format_type, advantages, examples in formats:
        print(f"{format_type:15} {advantages:25} 示例: {examples}")

demonstrate_serialization_concepts()

def compare_serialization_methods():
    """
    比较Python中的不同序列化方法
    """
    methods = [
        {
            'name': 'pickle',
            'type': '内置模块',
            '格式': '二进制',
            '优点': '支持几乎所有Python对象，使用简单',
            '缺点': 'Python特有，安全风险，版本兼容性问题',
            '适用场景': 'Python内部数据持久化，进程间通信'
        },
        {
            'name': 'json',
            'type': '内置模块',
            '格式': '文本',
            '优点': '跨语言支持，可读性好，广泛支持',
            '缺点': '不支持复杂Python对象，性能一般',
            '适用场景': 'Web API，配置文件，跨语言数据交换'
        },
        {
            'name': 'marshal',
            'type': '内置模块',
            '格式': '二进制',
            '优点': '高性能，用于Python字节码序列化',
            '缺点': '不保证跨版本兼容，不推荐一般使用',
            '适用场景': '.pyc文件，Python内部使用'
        },
        {
            'name': 'shelve',
            'type': '内置模块',
            '格式': '基于pickle的数据库',
            '优点': '类似字典的持久化存储接口',
            '缺点': '依赖pickle，有相同限制',
            '适用场景': '简单的键值对持久化'
        },
        {
            'name': '第三方库',
            'type': '多种选择',
            '格式': '多样',
            '优点': '专业功能，更好性能，更多特性',
            '缺点': '需要额外依赖',
            '适用场景': '高性能需求，特殊格式要求'
        }
    ]
    
    print("=== Python序列化方案比较 ===")
    for method in methods:
        print(f"\n{method['name']:10} ({method['type']}):")
        print(f"  格式: {method['格式']}")
        print(f"  优点: {method['优点']}")
        print(f"  缺点: {method['缺点']}")
        print(f"  场景: {method['适用场景']}")

compare_serialization_methods()

import pickle
import os

class AdvancedPickleDemo:
    """
    高级pickle功能演示
    """
    
    def __init__(self):
        self.serialized_data = None
    
    def demonstrate_basic_pickle(self):
        """演示基本pickle用法"""
        print("=== 基本pickle序列化 ===")
        
        # 创建复杂对象
        complex_object = {
            'string': 'Hello, 世界!',
            'number': 42,
            'list': [1, 2.5, 'three'],
            'tuple': (1, 2, 3),
            'set': {1, 2, 3},
            'dict': {'key': 'value'},
            'function': lambda x: x * 2,
            'none': None,
            'bool': True
        }
        
        # 序列化
        try:
            serialized = pickle.dumps(complex_object, protocol=pickle.HIGHEST_PROTOCOL)
            print(f"序列化大小: {len(serialized)} 字节")
            print(f"序列化数据 (前100字节): {serialized[:100]}...")
            
            # 反序列化
            deserialized = pickle.loads(serialized)
            print(f"反序列化成功: {type(deserialized)}")
            print(f"数据相等性: {deserialized == complex_object}")
            
            self.serialized_data = serialized
            return True
            
        except Exception as e:
            print(f"pickle错误: {e}")
            return False
    
    def demonstrate_file_operations(self):
        """演示文件序列化操作"""
        if not self.serialized_data:
            print("没有序列化数据")
            return False
        
        filename = 'pickle_demo.dat'
        
        try:
            # 写入文件
            with open(filename, 'wb') as f:
                pickle.dump(self.serialized_data, f, protocol=pickle.HIGHEST_PROTOCOL)
            print(f"数据已写入: {filename}")
            
            # 读取文件
            with open(filename, 'rb') as f:
                loaded_data = pickle.load(f)
            
            print(f"从文件加载数据大小: {len(loaded_data)} 字节")
            print(f"数据一致性: {loaded_data == self.serialized_data}")
            
            # 清理
            os.remove(filename)
            return True
            
        except Exception as e:
            print(f"文件操作错误: {e}")
            if os.path.exists(filename):
                os.remove(filename)
            return False
    
    def demonstrate_custom_classes(self):
        """演示自定义类的序列化"""
        print("=== 自定义类序列化 ===")
        
        class CustomClass:
            def __init__(self, name, value, items=None):
                self.name = name
                self.value = value
                self.items = items or []
                self._private_attr = "secret"
            
            def add_item(self, item):
                self.items.append(item)
            
            def __eq__(self, other):
                if not isinstance(other, CustomClass):
                    return False
                return (self.name == other.name and 
                        self.value == other.value and 
                        self.items == other.items)
            
            def __repr__(self):
                return f"CustomClass(name={self.name!r}, value={self.value!r})"
        
        # 创建实例
        original = CustomClass("test", 123)
        original.add_item("first")
        original.add_item("second")
        
        # 序列化
        try:
            serialized = pickle.dumps(original)
            print(f"自定义类序列化大小: {len(serialized)} 字节")
            
            # 反序列化
            reconstructed = pickle.loads(serialized)
            print(f"反序列化对象: {reconstructed}")
            print(f"对象相等性: {original == reconstructed}")
            print(f"类型一致性: {type(original) == type(reconstructed)}")
            
            return True
            
        except Exception as e:
            print(f"自定义类序列化错误: {e}")
            return False
    
    def demonstrate_protocols(self):
        """演示不同pickle协议"""
        print("=== pickle协议比较 ===")
        
        test_data = {
            'simple': 'string data',
            'complex': list(range(1000)),
            'nested': {'level1': {'level2': {'level3': 'deep'}}}
        }
        
        protocols = [
            (0, 'ASCII协议，可读但体积大'),
            (1, '旧二进制协议'),
            (2, 'Python 2.3+ 二进制协议'),
            (3, 'Python 3.0+ 二进制协议'),
            (4, 'Python 3.4+ 支持更大对象'),
            (5, 'Python 3.8+ 支持内存优化')
        ]
        
        for protocol_num, description in protocols:
            try:
                # 序列化
                data = pickle.dumps(test_data, protocol=protocol_num)
                size = len(data)
                
                # 反序列化
                reconstructed = pickle.loads(data)
                success = test_data == reconstructed
                
                print(f"协议 {protocol_num}: {size:6} 字节 - {description} - {'成功' if success else '失败'}")
                
            except Exception as e:
                print(f"协议 {protocol_num} 错误: {e}")

# 使用示例
def demo_pickle_features():
    """pickle功能演示"""
    demo = AdvancedPickleDemo()
    
    # 演示各种功能
    demo.demonstrate_basic_pickle()
    print()
    demo.demonstrate_file_operations()
    print()
    demo.demonstrate_custom_classes()
    print()
    demo.demonstrate_protocols()

demo_pickle_features()

import json
from datetime import datetime, date
from decimal import Decimal
from enum import Enum

class JSONAdvancedDemo:
    """
    JSON高级功能演示
    """
    
    def __init__(self):
        self.complex_data = {
            'string': 'Hello, JSON!',
            'number': 42.5,
            'boolean': True,
            'null_value': None,
            'array': [1, 'two', 3.0],
            'object': {'nested': 'value'},
            'timestamp': datetime.now(),
            'date': date.today(),
            'decimal': Decimal('123.456'),
            'set_data': {1, 2, 3}  # 集合需要特殊处理
        }
    
    def demonstrate_basic_json(self):
        """演示基本JSON序列化"""
        print("=== 基本JSON序列化 ===")
        
        try:
            # 基本序列化
            json_str = json.dumps(self.complex_data, indent=2)
            print(f"JSON字符串长度: {len(json_str)} 字符")
            print("JSON内容:")
            print(json_str[:200] + "..." if len(json_str) > 200 else json_str)
            
            # 反序列化
            parsed = json.loads(json_str)
            print(f"反序列化类型: {type(parsed)}")
            print(f"基本数据一致性: {parsed['string'] == self.complex_data['string']}")
            
            return True
            
        except Exception as e:
            print(f"JSON错误: {e}")
            return False
    
    def demonstrate_custom_serialization(self):
        """演示自定义序列化器"""
        print("=== 自定义JSON序列化 ===")
        
        # 自定义编码器
        class CustomEncoder(json.JSONEncoder):
            def default(self, obj):
                if isinstance(obj, datetime):
                    return obj.isoformat()
                elif isinstance(obj, date):
                    return obj.strftime('%Y-%m-%d')
                elif isinstance(obj, Decimal):
                    return float(obj)
                elif isinstance(obj, set):
                    return list(obj)
                elif isinstance(obj, Enum):
                    return obj.value
                # 让基类处理其他类型
                return super().default(obj)
        
        # 使用自定义编码器
        try:
            custom_json = json.dumps(self.complex_data, cls=CustomEncoder, indent=2)
            print("自定义序列化结果:")
            print(custom_json[:300] + "..." if len(custom_json) > 300 else custom_json)
            
            # 自定义解码器
            def custom_decoder(dct):
                # 可以在这里添加特殊处理逻辑
                return dct
            
            parsed_custom = json.loads(custom_json, object_hook=custom_decoder)
            print(f"自定义反序列化成功: {type(parsed_custom)}")
            
            return True
            
        except Exception as e:
            print(f"自定义序列化错误: {e}")
            return False
    
    def demonstrate_json_performance(self):
        """演示JSON性能考虑"""
        print("=== JSON性能优化 ===")
        
        # 创建大型测试数据
        large_data = {
            'users': [{'id': i, 'name': f'user_{i}', 'data': list(range(100))} 
                     for i in range(1000)],
            'metadata': {'timestamp': datetime.now().isoformat()}
        }
        
        # 比较不同选项的性能
        options = [
            ('默认', {}),
            ('无缩进', {'indent': None}),
            ('分隔符优化', {'separators': (',', ':')}),
            ('ASCII编码', {'ensure_ascii': True}),
            ('性能模式', {'indent': None, 'separators': (',', ':')})
        ]
        
        import time
        
        for name, kwargs in options:
            start_time = time.time()
            
            try:
                # 序列化
                json_data = json.dumps(large_data, **kwargs)
                serialize_time = time.time() - start_time
                
                # 反序列化
                start_time = time.time()
                parsed = json.loads(json_data)
                deserialize_time = time.time() - start_time
                
                total_time = serialize_time + deserialize_time
                size = len(json_data)
                
                print(f"{name:15} 大小: {size:6} 字节, 序列化: {serialize_time:.4f}s, 反序列化: {deserialize_time:.4f}s, 总计: {total_time:.4f}s")
                
            except Exception as e:
                print(f"{name} 错误: {e}")
    
    def demonstrate_json_schema(self):
        """演示JSON Schema验证"""
        print("=== JSON Schema验证 ===")
        
        # 简单的schema验证示例
        schema = {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "number", "minimum": 0},
                "email": {"type": "string", "format": "email"},
                "tags": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["name", "age"]
        }
        
        # 测试数据
        test_cases = [
            {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'},  # 有效
            {'name': 'Bob', 'age': -5},  # 年龄无效
            {'age': 25},  # 缺少必填字段
            {'name': 123, 'age': 25}  # 名称类型错误
        ]
        
        # 简单验证函数（实际应用中应该使用jsonschema库）
        def simple_validate(data, schema):
            errors = []
            
            # 检查必填字段
            for field in schema.get('required', []):
                if field not in data:
                    errors.append(f"缺少必填字段: {field}")
            
            # 检查字段类型
            for field, value in data.items():
                if field in schema.get('properties', {}):
                    field_schema = schema['properties'][field]
                    expected_type = field_schema.get('type')
                    
                    if expected_type == 'string' and not isinstance(value, str):
                        errors.append(f"字段 {field} 应该是字符串类型")
                    elif expected_type == 'number' and not isinstance(value, (int, float)):
                        errors.append(f"字段 {field} 应该是数字类型")
                    elif expected_type == 'array' and not isinstance(value, list):
                        errors.append(f"字段 {field} 应该是数组类型")
            
            return len(errors) == 0, errors
        
        # 测试验证
        for i, test_data in enumerate(test_cases):
            is_valid, errors = simple_validate(test_data, schema)
            status = "有效" if is_valid else "无效"
            print(f"测试用例 {i+1}: {status}")
            if errors:
                for error in errors:
                    print(f"  - {error}")

# 使用示例
def demo_json_features():
    """JSON功能演示"""
    demo = JSONAdvancedDemo()
    
    demo.demonstrate_basic_json()
    print()
    demo.demonstrate_custom_serialization()
    print()
    demo.demonstrate_json_performance()
    print()
    demo.demonstrate_json_schema()

demo_json_features()

class CustomSerializationFramework:
    """
    自定义序列化框架
    """
    
    def __init__(self):
        self.serializers = {}
        self.deserializers = {}
        self._register_builtin_types()
    
    def _register_builtin_types(self):
        """注册内置类型处理器"""
        # 基本类型
        self.register_serializer(str, lambda x: ('str', x))
        self.register_serializer(int, lambda x: ('int', x))
        self.register_serializer(float, lambda x: ('float', x))
        self.register_serializer(bool, lambda x: ('bool', x))
        self.register_serializer(type(None), lambda x: ('none', None))
        
        # 容器类型
        self.register_serializer(list, lambda x: ('list', [self.serialize(item) for item in x]))
        self.register_serializer(dict, lambda x: ('dict', {k: self.serialize(v) for k, v in x.items()}))
        self.register_serializer(tuple, lambda x: ('tuple', [self.serialize(item) for item in x]))
        self.register_serializer(set, lambda x: ('set', [self.serialize(item) for item in x]))
        
        # 注册反序列化器
        self.register_deserializer('str', lambda x: x)
        self.register_deserializer('int', int)
        self.register_deserializer('float', float)
        self.register_deserializer('bool', bool)
        self.register_deserializer('none', lambda x: None)
        self.register_deserializer('list', lambda x: [self.deserialize(item) for item in x])
        self.register_deserializer('dict', lambda x: {k: self.deserialize(v) for k, v in x.items()})
        self.register_deserializer('tuple', lambda x: tuple(self.deserialize(item) for item in x))
        self.register_deserializer('set', lambda x: set(self.deserialize(item) for item in x))
    
    def register_serializer(self, data_type, serializer_func):
        """注册序列化器"""
        type_name = data_type.__name__ if hasattr(data_type, '__name__') else str(data_type)
        self.serializers[type_name] = serializer_func
    
    def register_deserializer(self, type_name, deserializer_func):
        """注册反序列化器"""
        self.deserializers[type_name] = deserializer_func
    
    def serialize(self, obj):
        """序列化对象"""
        obj_type = type(obj)
        type_name = obj_type.__name__
        
        if type_name in self.serializers:
            return self.serializers[type_name](obj)
        else:
            # 尝试处理未知类型
            try:
                # 对于自定义对象，使用字典表示
                if hasattr(obj, '__dict__'):
                    return ('object', {
                        '__class__': obj.__class__.__name__,
                        '__module__': obj.__module__,
                        'data': {k: self.serialize(v) for k, v in obj.__dict__.items()}
                    })
                else:
                    raise ValueError(f"无法序列化类型: {type_name}")
            except Exception as e:
                raise ValueError(f"序列化错误: {e}")
    
    def deserialize(self, serialized_data):
        """反序列化数据"""
        if not isinstance(serialized_data, (list, tuple)) or len(serialized_data) != 2:
            raise ValueError("无效的序列化数据格式")
        
        type_name, data = serialized_data
        
        if type_name in self.deserializers:
            return self.deserializers[type_name](data)
        elif type_name == 'object':
            # 处理自定义对象
            class_name = data['__class__']
            module_name = data['__module__']
            obj_data = data['data']
            
            # 动态导入模块（生产环境需要更安全的方法）
            try:
                module = __import__(module_name, fromlist=[class_name])
                obj_class = getattr(module, class_name)
                
                # 创建对象实例
                instance = obj_class.__new__(obj_class)
                
                # 恢复属性
                for attr_name, attr_value in obj_data.items():
                    setattr(instance, attr_name, self.deserialize(attr_value))
                
                return instance
            except Exception as e:
                raise ValueError(f"反序列化对象错误: {e}")
        else:
            raise ValueError(f"未知的类型标识: {type_name}")
    
    def to_json_compatible(self, obj):
        """转换为JSON兼容格式"""
        serialized = self.serialize(obj)
        return serialized
    
    def from_json_compatible(self, data):
        """从JSON兼容格式恢复"""
        return self.deserialize(data)

# 使用示例
def demo_custom_serialization():
    """自定义序列化演示"""
    print("=== 自定义序列化框架 ===")
    
    class TestClass:
        def __init__(self, name, value, items=None):
            self.name = name
            self.value = value
            self.items = items or []
            self._private = "private_data"
        
        def __eq__(self, other):
            if not isinstance(other, TestClass):
                return False
            return (self.name == other.name and 
                    self.value == other.value and 
                    self.items == other.items)
        
        def __repr__(self):
            return f"TestClass({self.name!r}, {self.value!r})"
    
    # 创建框架实例
    framework = CustomSerializationFramework()
    
    # 注册自定义类型
    framework.register_serializer(TestClass, 
        lambda x: ('object', {
            '__class__': 'TestClass',
            '__module__': '__main__',
            'data': {
                'name': framework.serialize(x.name),
                'value': framework.serialize(x.value),
                'items': framework.serialize(x.items)
            }
        }))
    
    # 测试数据
    test_obj = TestClass("test", 42, [1, "two", 3.0])
    nested_data = {
        'string': 'hello',
        'number': 123,
        'object': test_obj,
        'list': [test_obj, test_obj],
        'set': {1, 2, 3}
    }
    
    try:
        # 序列化
        serialized = framework.serialize(nested_data)
        print(f"序列化结果: {serialized}")
        
        # 转换为JSON兼容格式
        json_compatible = framework.to_json_compatible(nested_data)
        print(f"JSON兼容格式: {json_compatible}")
        
        # 反序列化
        deserialized = framework.deserialize(serialized)
        print(f"反序列化成功: {type(deserialized)}")
        
        # 验证
        print(f"数据一致性: {nested_data['string'] == deserialized['string']}")
        print(f"对象一致性: {nested_data['object'] == deserialized['object']}")
        
    except Exception as e:
        print(f"自定义序列化错误: {e}")

demo_custom_serialization()

import msgpack
import umsgpack
import rapidjson
import orjson

class HighPerformanceSerializer:
    """
    高性能序列化方案比较
    """
    
    def __init__(self):
        self.test_data = self._create_test_data()
    
    def _create_test_data(self):
        """创建测试数据"""
        return {
            'users': [
                {
                    'id': i,
                    'name': f'user_{i}',
                    'email': f'user{i}@example.com',
                    'profile': {
                        'age': 20 + (i % 40),
                        'score': 100.0 - (i * 0.1),
                        'tags': ['tag1', 'tag2', 'tag3'],
                        'active': i % 2 == 0
                    },
                    'history': list(range(50)),
                    'metadata': {
                        'created': '2023-01-01',
                        'updated': '2023-12-31',
                        'flags': [True, False, True]
                    }
                }
                for i in range(1000)
            ],
            'metadata': {
                'timestamp': datetime.now().isoformat(),
                'version': '1.0.0',
                'count': 1000,
                'stats': {
                    'min_age': 20,
                    'max_age': 59,
                    'avg_score': 50.0
                }
            }
        }
    
    def benchmark_serializers(self):
        """性能基准测试"""
        print("=== 序列化性能基准测试 ===")
        
        libraries = [
            ('json', lambda x: json.dumps(x), lambda x: json.loads(x)),
            ('rapidjson', lambda x: rapidjson.dumps(x), lambda x: rapidjson.loads(x)),
            ('orjson', lambda x: orjson.dumps(x), lambda x: orjson.loads(x)),
            ('msgpack', lambda x: msgpack.packb(x), lambda x: msgpack.unpackb(x)),
            ('umsgpack', lambda x: umsgpack.packb(x), lambda x: umsgpack.unpackb(x)),
            ('pickle', lambda x: pickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL), 
             lambda x: pickle.loads(x))
        ]
        
        import time
        results = []
        
        for name, serializer, deserializer in libraries:
            try:
                # 预热
                if name != 'pickle':  # pickle不需要预热
                    serializer(self.test_data)
                    deserializer(serializer(self.test_data))
                
                # 序列化测试
                serialize_times = []
                for _ in range(5):
                    start_time = time.time()
                    serialized_data = serializer(self.test_data)
                    serialize_times.append(time.time() - start_time)
                
                avg_serialize = sum(serialize_times) / len(serialize_times)
                serialized_size = len(serialized_data)
                
                # 反序列化测试
                deserialize_times = []
                for _ in range(5):
                    start_time = time.time()
                    deserialized_data = deserializer(serialized_data)
                    deserialize_times.append(time.time() - start_time)
                
                avg_deserialize = sum(deserialize_times) / len(deserialize_times)
                
                # 验证数据一致性
                if name != 'pickle':  # pickle可能无法直接比较
                    is_valid = deserialized_data == self.test_data
                else:
                    is_valid = True
                
                results.append({
                    'name': name,
                    'serialize_time': avg_serialize,
                    'deserialize_time': avg_deserialize,
                    'total_time': avg_serialize + avg_deserialize,
                    'size': serialized_size,
                    'valid': is_valid
                })
                
            except Exception as e:
                print(f"{name} 测试失败: {e}")
                results.append({
                    'name': name,
                    'error': str(e)
                })
        
        # 显示结果
        print(f"{'库名':<12} {'序列化':<8} {'反序列化':<8} {'总计':<8} {'大小':<8} {'验证'}")
        print("-" * 60)
        
        for result in sorted(results, key=lambda x: x.get('total_time', float('inf'))):
            if 'error' in result:
                print(f"{result['name']:<12} 错误: {result['error']}")
            else:
                print(f"{result['name']:<12} {result['serialize_time']:.6f} {result['deserialize_time']:.6f} "
                      f"{result['total_time']:.6f} {result['size']:<8} {'✓' if result['valid'] else '✗'}")
        
        return results
    
    def demonstrate_msgpack_features(self):
        """演示MessagePack特性"""
        print("=== MessagePack特性演示 ===")
        
        try:
            # 序列化
            packed = msgpack.packb(self.test_data)
            print(f"MessagePack大小: {len(packed)} 字节")
            print(f"JSON大小: {len(json.dumps(self.test_data))} 字节")
            print(f"压缩比: {len(json.dumps(self.test_data)) / len(packed):.2f}x")
            
            # 反序列化
            unpacked = msgpack.unpackb(packed)
            print(f"反序列化成功: {type(unpacked)}")
            print(f"数据一致性: {unpacked['users'][0]['name'] == self.test_data['users'][0]['name']}")
            
            # 显示部分二进制数据
            print(f"二进制数据 (前50字节): {packed[:50].hex(' ')}...")
            
        except Exception as e:
            print(f"MessagePack错误: {e}")
    
    def demonstrate_orjson_features(self):
        """演示orjson特性"""
        print("=== orjson特性演示 ===")
        
        try:
            # orjson支持更多数据类型
            extended_data = self.test_data.copy()
            extended_data['datetime'] = datetime.now()
            extended_data['date'] = date.today()
            extended_data['decimal'] = Decimal('123.456')
            extended_data['uuid'] = '6ba7b810-9dad-11d1-80b4-00c04fd430c8'
            
            # 序列化
            serialized = orjson.dumps(extended_data)
            print(f"orjson大小: {len(serialized)} 字节")
            
            # 反序列化
            deserialized = orjson.loads(serialized)
            print(f"反序列化成功: {type(deserialized)}")
            
            # orjson特性
            print("orjson支持:")
            print("  - 内置datetime支持")
            print("  - 高性能C实现")
            print("  - 无GIL限制")
            print("  - 内存效率高")
            
        except Exception as e:
            print(f"orjson错误: {e}")

# 使用示例
def demo_performance_serialization():
    """性能序列化演示"""
    perf = HighPerformanceSerializer()
    
    # 运行性能测试
    results = perf.benchmark_serializers()
    print()
    
    # 演示特定库特性
    perf.demonstrate_msgpack_features()
    print()
    perf.demonstrate_orjson_features()

demo_performance_serialization()

class SecureSerialization:
    """
    安全序列化实践
    """
    
    def demonstrate_pickle_security_risks(self):
        """演示pickle安全风险"""
        print("=== pickle安全风险演示 ===")
        
        # 危险的pickle数据
        malicious_code = """
import os
os.system('echo "危险操作被执行"')
"""
        
        # 创建恶意pickle数据
        class MaliciousPayload:
            def __reduce__(self):
                return (eval, (malicious_code,))
        
        try:
            # 序列化恶意载荷
            malicious_pickle = pickle.dumps(MaliciousPayload())
            print(f"恶意pickle数据创建: {len(malicious_pickle)} 字节")
            
            # 警告：不要在实际环境中执行以下代码
            print("警告: 以下操作可能危险，仅在受控环境中演示")
            
            # 演示反序列化风险（注释掉实际执行）
            # result = pickle.loads(malicious_pickle)
            # print(f"恶意代码执行结果: {result}")
            
            print("✅ 实际执行已禁用，仅用于演示风险")
            
        except Exception as e:
            print(f"恶意代码演示错误: {e}")
    
    def demonstrate_safe_alternatives(self):
        """演示安全替代方案"""
        print("=== 安全序列化替代方案 ===")
        
        # 1. 使用JSON进行安全数据交换
        safe_data = {
            'name': 'safe_data',
            'value': 42,
            'items': ['a', 'b', 'c']
        }
        
        json_str = json.dumps(safe_data)
        json_parsed = json.loads(json_str)
        print(f"JSON安全序列化: {json_str[:50]}...")
        print(f"JSON安全反序列化: {type(json_parsed)}")
        
        # 2. 使用白名单控制pickle
        class SafeUnpickler(pickle.Unpickler):
            def find_class(self, module, name):
                # 只允许安全的模块和类
                safe_modules = {'builtins', '__main__', 'datetime'}
                safe_classes = {'str', 'int', 'float', 'list', 'dict', 'tuple'}
                
                if module not in safe_modules:
                    raise pickle.UnpicklingError(f"不安全的模块: {module}")
                if name not in safe_classes:
                    raise pickle.UnpicklingError(f"不安全的类: {name}")
                
                return super().find_class(module, name)
        
        # 测试安全unpickler
        safe_data = ['safe', 'data', 123]
        safe_pickle = pickle.dumps(safe_data)
        
        try:
            # 使用安全unpickler
            safe_result = SafeUnpickler(io.BytesIO(safe_pickle)).load()
            print(f"安全unpickler结果: {safe_result}")
        except Exception as e:
            print(f"安全unpickler错误: {e}")
        
        # 3. 数据验证和清洗
        def sanitize_data(data):
            """数据清洗函数"""
            if isinstance(data, dict):
                return {k: sanitize_data(v) for k, v in data.items()}
            elif isinstance(data, list):
                return [sanitize_data(item) for item in data]
            elif isinstance(data, (str, int, float, bool)):
                return data
            else:
                # 拒绝不安全的类型
                raise ValueError(f"不安全的数据类型: {type(data)}")
        
        try:
            cleaned_data = sanitize_data(safe_data)
            print(f"数据清洗成功: {cleaned_data}")
        except Exception as e:
            print(f"数据清洗错误: {e}")
    
    def demonstrate_encrypted_serialization(self):
        """演示加密序列化"""
        print("=== 加密序列化演示 ===")
        
        from cryptography.fernet import Fernet
        
        # 生成加密密钥
        key = Fernet.generate_key()
        cipher = Fernet(key)
        
        # 要加密的数据
        sensitive_data = {
            'username': 'admin',
            'password': 'secret123',  # 实际应用中应该使用哈希
            'token': 'abcdef123456'
        }
        
        try:
            # 序列化后加密
            serialized = pickle.dumps(sensitive_data)
            encrypted = cipher.encrypt(serialized)
            
            print(f"原始数据大小: {len(serialized)} 字节")
            print(f"加密数据大小: {len(encrypted)} 字节")
            print(f"加密数据: {encrypted[:30]}...")
            
            # 解密和反序列化
            decrypted = cipher.decrypt(encrypted)
            deserialized = pickle.loads(decrypted)
            
            print(f"解密成功: {deserialized['username']}")
            print(f"数据完整性: {deserialized == sensitive_data}")
            
        except Exception as e:
            print(f"加密序列化错误: {e}")
    
    def demonstrate_signing_data(self):
        """演示数据签名"""
        print("=== 数据签名演示 ===")
        
        import hmac
        import hashlib
        
        # 共享密钥（实际应用中应该安全存储）
        secret_key = b'my_secret_key'
        
        data_to_sign = {'important': 'data', 'timestamp': time.time()}
        serialized_data = json.dumps(data_to_sign).encode('utf-8')
        
        # 创建签名
        signature = hmac.new(secret_key, serialized_data, hashlib.sha256).hexdigest()
        
        # 组合数据和签名
        signed_package = {
            'data': data_to_sign,
            'signature': signature
        }
        
        print(f"签名数据包: {signed_package}")
        
        # 验证签名
        def verify_signature(data, received_signature):
            """验证数据签名"""
            serialized = json.dumps(data).encode('utf-8')
            expected_signature = hmac.new(secret_key, serialized_data, hashlib.sha256).hexdigest()
            return hmac.compare_digest(expected_signature, received_signature)
        
        # 测试验证
        is_valid = verify_signature(data_to_sign, signature)
        print(f"签名验证: {'成功' if is_valid else '失败'}")
        
        # 测试篡改检测
        tampered_data = {'important': 'modified', 'timestamp': time.time()}
        is_tampered_valid = verify_signature(tampered_data, signature)
        print(f"篡改检测: {'检测到篡改' if not is_tampered_valid else '未检测到篡改'}")

# 使用示例
def demo_security_features():
    """安全特性演示"""
    security = SecureSerialization()
    
    security.demonstrate_pickle_security_risks()
    print()
    security.demonstrate_safe_alternatives()
    print()
    security.demonstrate_encrypted_serialization()
    print()
    security.demonstrate_signing_data()

demo_security_features()

class ConfigurationManager:
    """
    基于序列化的配置管理系统
    """
    
    def __init__(self, config_file='config.json'):
        self.config_file = config_file
        self.config_data = {}
        self.default_config = {
            'app': {
                'name': 'My Application',
                'version': '1.0.0',
                'debug': False
            },
            'database': {
                'host': 'localhost',
                'port': 5432,
                'name': 'mydb',
                'user': 'admin'
            },
            'logging': {
                'level': 'INFO',
                'file': 'app.log',
                'max_size': 10485760
            }
        }
    
    def load_configuration(self, file_format='json'):
        """
        加载配置文件
        """
        if not os.path.exists(self.config_file):
            print(f"配置文件不存在，使用默认配置: {self.config_file}")
            self.config_data = self.default_config.copy()
            return True
        
        try:
            with open(self.config_file, 'rb') as f:
                if file_format == 'json':
                    self.config_data = json.load(f)
                elif file_format == 'pickle':
                    self.config_data = pickle.load(f)
                elif file_format == 'yaml':
                    import yaml
                    self.config_data = yaml.safe_load(f)
                else:
                    raise ValueError(f"不支持的格式: {file_format}")
            
            print(f"配置文件加载成功: {self.config_file}")
            return True
            
        except Exception as e:
            print(f"配置文件加载错误: {e}")
            self.config_data = self.default_config.copy()
            return False
    
    def save_configuration(self, file_format='json'):
        """
        保存配置文件
        """
        try:
            # 确保目录存在
            os.makedirs(os.path.dirname(os.path.abspath(self.config_file)), exist_ok=True)
            
            with open(self.config_file, 'wb') as f:
                if file_format == 'json':
                    f.write(json.dumps(self.config_data, indent=2).encode('utf-8'))
                elif file_format == 'pickle':
                    pickle.dump(self.config_data, f, protocol=pickle.HIGHEST_PROTOCOL)
                elif file_format == 'yaml':
                    import yaml
                    yaml.dump(self.config_data, f, default_flow_style=False)
                else:
                    raise ValueError(f"不支持的格式: {file_format}")
            
            print(f"配置文件保存成功: {self.config_file}")
            return True
            
        except Exception as e:
            print(f"配置文件保存错误: {e}")
            return False
    
    def get_config_value(self, key_path, default=None):
        """
        获取配置值
        """
        try:
            value = self.config_data
            for key in key_path.split('.'):
                value = value[key]
            return value
        except (KeyError, TypeError):
            return default
    
    def set_config_value(self, key_path, value):
        """
        设置配置值
        """
        keys = key_path.split('.')
        current_level = self.config_data
        
        for key in keys[:-1]:
            if key not in current_level:
                current_level[key] = {}
            current_level = current_level[key]
        
        current_level[keys[-1]] = value
        return True
    
    def validate_configuration(self, schema=None):
        """
        验证配置有效性
        """
        # 简单的验证逻辑
        required_keys = [
            'app.name',
            'app.version',
            'database.host',
            'database.port'
        ]
        
        errors = []
        for key in required_keys:
            if self.get_config_value(key) is None:
                errors.append(f"缺少必填配置项: {key}")
        
        # 验证端口范围
        db_port = self.get_config_value('database.port')
        if db_port and not (0 < db_port < 65536):
            errors.append(f"数据库端口无效: {db_port}")
        
        # 验证日志级别
        log_level = self.get_config_value('logging.level')
        valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
        if log_level and log_level.upper() not in valid_levels:
            errors.append(f"无效的日志级别: {log_level}")
        
        if errors:
            print("配置验证失败:")
            for error in errors:
                print(f"  - {error}")
            return False
        else:
            print("配置验证成功")
            return True
    
    def migrate_configuration(self, old_version, new_version):
        """
        配置迁移
        """
        print(f"迁移配置从 {old_version} 到 {new_version}")
        
        # 简单的迁移逻辑
        migration_scripts = {
            '1.0.0_to_1.1.0': lambda config: config.update({'new_feature': 'enabled'}),
            '1.1.0_to_1.2.0': lambda config: config.pop('deprecated_setting', None)
        }
        
        migration_key = f"{old_version}_to_{new_version}"
        if migration_key in migration_scripts:
            migration_scripts[migration_key](self.config_data)
            print(f"迁移完成: {migration_key}")
            return True
        else:
            print(f"没有找到迁移脚本: {migration_key}")
            return False

# 使用示例
def demo_configuration_manager():
    """配置管理器演示"""
    print("=== 配置管理系统 ===")
    
    manager = ConfigurationManager('demo_config.json')
    
    # 加载配置
    manager.load_configuration()
    
    # 修改配置
    manager.set_config_value('app.debug', True)
    manager.set_config_value('database.port', 3306)
    manager.set_config_value('new_setting.nested.value', 'test')
    
    # 验证配置
    manager.validate_configuration()
    
    # 保存配置
    manager.save_configuration()
    
    # 读取配置值
    app_name = manager.get_config_value('app.name')
    db_port = manager.get_config_value('database.port')
    print(f"应用名称: {app_name}")
    print(f"数据库端口: {db_port}")
    
    # 清理
    if os.path.exists('demo_config.json'):
        os.remove('demo_config.json')

demo_configuration_manager()

class DistributedTaskQueue:
    """
    基于序列化的分布式任务队列
    """
    
    def __init__(self, queue_name='default', serializer='pickle'):
        self.queue_name = queue_name
        self.serializer = serializer
        self.tasks = []
        
        # 创建序列化器
        self.serializers = {
            'pickle': {
                'dump': pickle.dumps,
                'load': pickle.loads
            },
            'json': {
                'dump': lambda x: json.dumps(x).encode('utf-8'),
                'load': lambda x: json.loads(x.decode('utf-8'))
            },
            'msgpack': {
                'dump': msgpack.packb,
                'load': msgpack.unpackb
            }
        }
    
    def enqueue_task(self, task_func, *args, **kwargs):
        """
        将任务加入队列
        """
        task_id = f"task_{len(self.tasks)}_{int(time.time())}"
        
        task_data = {
            'id': task_id,
            'function': task_func.__name__ if callable(task_func) else str(task_func),
            'module': task_func.__module__ if hasattr(task_func, '__module__') else '__main__',
            'args': args,
            'kwargs': kwargs,
            'created_at': time.time(),
            'status': 'pending'
        }
        
        self.tasks.append(task_data)
        print(f"任务已加入队列: {task_id}")
        return task_id
    
    def serialize_queue(self, filename=None):
        """
        序列化任务队列
        """
        if self.serializer not in self.serializers:
            raise ValueError(f"不支持的序列化器: {self.serializer}")
        
        serializer = self.serializers[self.serializer]['dump']
        serialized_data = serializer(self.tasks)
        
        if filename:
            with open(filename, 'wb') as f:
                f.write(serialized_data)
            print(f"队列已序列化到文件: {filename}")
        
        return serialized_data
    
    def deserialize_queue(self, data=None, filename=None):
        """
        反序列化任务队列
        """
        if self.serializer not in self.serializers:
            raise ValueError(f"不支持的序列化器: {self.serializer}")
        
        deserializer = self.serializers[self.serializer]['load']
        
        if filename:
            with open(filename, 'rb') as f:
                data = f.read()
        
        if data:
            self.tasks = deserializer(data)
            print(f"队列已从{'文件' if filename else '数据'}加载: {len(self.tasks)} 个任务")
            return True
        
        return False
    
    def process_tasks(self, max_tasks=None):
        """
        处理任务
        """
        processed = 0
        max_tasks = max_tasks or len(self.tasks)
        
        for i, task in enumerate(self.tasks[:max_tasks]):
            if task['status'] == 'pending':
                print(f"处理任务 {i+1}: {task['id']}")
                try:
                    # 模拟任务处理
                    result = f"处理结果: {task['function']}({len(task['args'])} 参数)"
                    task['status'] = 'completed'
                    task['result'] = result
                    task['completed_at'] = time.time()
                    processed += 1
                    
                    print(f"任务完成: {result}")
                    
                except Exception as e:
                    task['status'] = 'failed'
                    task['error'] = str(e)
                    print(f"任务失败: {e}")
        
        return processed
    
    def get_queue_stats(self):
        """
        获取队列统计
        """
        stats = {
            'total_tasks': len(self.tasks),
            'pending': sum(1 for t in self.tasks if t['status'] == 'pending'),
            'completed': sum(1 for t in self.tasks if t['status'] == 'completed'),
            'failed': sum(1 for t in self.tasks if t['status'] == 'failed'),
            'oldest_task': min((t['created_at'] for t in self.tasks), default=0),
            'newest_task': max((t['created_at'] for t in self.tasks), default=0)
        }
        
        print("队列统计:")
        for key, value in stats.items():
            print(f"  {key}: {value}")
        
        return stats
    
    def clear_queue(self, status=None):
        """
        清理队列
        """
        if status:
            self.tasks = [t for t in self.tasks if t['status'] != status]
            print(f"已清理状态为 {status} 的任务")
        else:
            self.tasks.clear()
            print("队列已清空")

# 使用示例
def demo_task_queue():
    """任务队列演示"""
    print("=== 分布式任务队列 ===")
    
    # 创建任务函数
    def process_data(data, multiplier=1):
        return f"处理了 {len(data)} 条数据，结果: {[x * multiplier for x in data]}"
    
    def send_email(to, subject, body):
        return f"发送邮件到 {to}: {subject}"
    
    def generate_report(format='pdf', pages=10):
        return f"生成 {format} 报告，{pages} 页"
    
    # 创建队列
    queue = DistributedTaskQueue(serializer='json')
    
    # 添加任务
    queue.enqueue_task(process_data, [1, 2, 3, 4, 5], multiplier=2)
    queue.enqueue_task(send_email, 'user@example.com', '重要通知', '请查收附件')
    queue.enqueue_task(generate_report, 'excel', 25)
    
    # 显示统计
    queue.get_queue_stats()
    
    # 序列化队列
    serialized = queue.serialize_queue('task_queue.json')
    print(f"序列化数据大小: {len(serialized)} 字节")
    
    # 处理任务
    processed = queue.process_tasks(2)
    print(f"处理了 {processed} 个任务")
    
    # 更新统计
    queue.get_queue_stats()
    
    # 清理
    queue.clear_queue('completed')
    queue.get_queue_stats()
    
    # 保存最终状态
    queue.serialize_queue('task_queue_final.json')
    
    # 清理文件
    for filename in ['task_queue.json', 'task_queue_final.json']:
        if os.path.exists(filename):
            os.remove(filename)

demo_task_queue()