def demonstrate_serialization_concepts():
    """
    演示序列化的核心概念和用途
    """
    concepts = {
        '序列化 (Serialization)': '將對象轉(zhuǎn)換為字節(jié)流或文本格式的過程',
        '反序列化 (Deserialization)': '從序列化格式重建原始對象的過程',
        '持久化 (Persistence)': '將對象狀態(tài)保存到持久存儲（如文件、數(shù)據(jù)庫）',
        '數(shù)據(jù)交換 (Data Exchange)': '在不同系統(tǒng)或進(jìn)程間傳輸對象數(shù)據(jù)',
        '深度復(fù)制 (Deep Copy)': '通過序列化/反序列化實(shí)現(xiàn)對象的深度復(fù)制',
        '狀態(tài)恢復(fù) (State Restoration)': '保存和恢復(fù)應(yīng)用程序狀態(tài)'
    }
    
    print("=== 序列化核心概念 ===")
    for term, definition in concepts.items():
        print(f"{term:20}: {definition}")
    
    # 序列化的常見格式
    formats = [
        ('二進(jìn)制格式', '緊湊高效，但不易讀', 'pickle, protobuf'),
        ('文本格式', '可讀性好，但體積較大', 'JSON, XML, YAML'),
        ('混合格式', '平衡可讀性和效率', 'MessagePack, BSON'),
        ('專用格式', '針對特定場景優(yōu)化', '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': '內(nèi)置模塊',
            '格式': '二進(jìn)制',
            '優(yōu)點(diǎn)': '支持幾乎所有Python對象，使用簡單',
            '缺點(diǎn)': 'Python特有，安全風(fēng)險(xiǎn)，版本兼容性問題',
            '適用場景': 'Python內(nèi)部數(shù)據(jù)持久化，進(jìn)程間通信'
        },
        {
            'name': 'json',
            'type': '內(nèi)置模塊',
            '格式': '文本',
            '優(yōu)點(diǎn)': '跨語言支持，可讀性好，廣泛支持',
            '缺點(diǎn)': '不支持復(fù)雜Python對象，性能一般',
            '適用場景': 'Web API，配置文件，跨語言數(shù)據(jù)交換'
        },
        {
            'name': 'marshal',
            'type': '內(nèi)置模塊',
            '格式': '二進(jìn)制',
            '優(yōu)點(diǎn)': '高性能，用于Python字節(jié)碼序列化',
            '缺點(diǎn)': '不保證跨版本兼容，不推薦一般使用',
            '適用場景': '.pyc文件，Python內(nèi)部使用'
        },
        {
            'name': 'shelve',
            'type': '內(nèi)置模塊',
            '格式': '基于pickle的數(shù)據(jù)庫',
            '優(yōu)點(diǎn)': '類似字典的持久化存儲接口',
            '缺點(diǎn)': '依賴pickle，有相同限制',
            '適用場景': '簡單的鍵值對持久化'
        },
        {
            'name': '第三方庫',
            'type': '多種選擇',
            '格式': '多樣',
            '優(yōu)點(diǎn)': '專業(yè)功能，更好性能，更多特性',
            '缺點(diǎn)': '需要額外依賴',
            '適用場景': '高性能需求，特殊格式要求'
        }
    ]
    
    print("=== Python序列化方案比較 ===")
    for method in methods:
        print(f"\n{method['name']:10} ({method['type']}):")
        print(f"  格式: {method['格式']}")
        print(f"  優(yōu)點(diǎn): {method['優(yōu)點(diǎn)']}")
        print(f"  缺點(diǎn): {method['缺點(diǎn)']}")
        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序列化 ===")
        
        # 創(chuàng)建復(fù)雜對象
        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)} 字節(jié)")
            print(f"序列化數(shù)據(jù) (前100字節(jié)): {serialized[:100]}...")
            
            # 反序列化
            deserialized = pickle.loads(serialized)
            print(f"反序列化成功: {type(deserialized)}")
            print(f"數(shù)據(jù)相等性: {deserialized == complex_object}")
            
            self.serialized_data = serialized
            return True
            
        except Exception as e:
            print(f"pickle錯(cuò)誤: {e}")
            return False
    
    def demonstrate_file_operations(self):
        """演示文件序列化操作"""
        if not self.serialized_data:
            print("沒有序列化數(shù)據(jù)")
            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"數(shù)據(jù)已寫入: {filename}")
            
            # 讀取文件
            with open(filename, 'rb') as f:
                loaded_data = pickle.load(f)
            
            print(f"從文件加載數(shù)據(jù)大小: {len(loaded_data)} 字節(jié)")
            print(f"數(shù)據(jù)一致性: {loaded_data == self.serialized_data}")
            
            # 清理
            os.remove(filename)
            return True
            
        except Exception as e:
            print(f"文件操作錯(cuò)誤: {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})"
        
        # 創(chuàng)建實(shí)例
        original = CustomClass("test", 123)
        original.add_item("first")
        original.add_item("second")
        
        # 序列化
        try:
            serialized = pickle.dumps(original)
            print(f"自定義類序列化大小: {len(serialized)} 字節(jié)")
            
            # 反序列化
            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"自定義類序列化錯(cuò)誤: {e}")
            return False
    
    def demonstrate_protocols(self):
        """演示不同pickle協(xié)議"""
        print("=== pickle協(xié)議比較 ===")
        
        test_data = {
            'simple': 'string data',
            'complex': list(range(1000)),
            'nested': {'level1': {'level2': {'level3': 'deep'}}}
        }
        
        protocols = [
            (0, 'ASCII協(xié)議，可讀但體積大'),
            (1, '舊二進(jìn)制協(xié)議'),
            (2, 'Python 2.3+ 二進(jìn)制協(xié)議'),
            (3, 'Python 3.0+ 二進(jìn)制協(xié)議'),
            (4, 'Python 3.4+ 支持更大對象'),
            (5, 'Python 3.8+ 支持內(nèi)存優(yōu)化')
        ]
        
        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"協(xié)議 {protocol_num}: {size:6} 字節(jié) - {description} - {'成功' if success else '失敗'}")
                
            except Exception as e:
                print(f"協(xié)議 {protocol_num} 錯(cuò)誤: {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內(nèi)容:")
            print(json_str[:200] + "..." if len(json_str) > 200 else json_str)
            
            # 反序列化
            parsed = json.loads(json_str)
            print(f"反序列化類型: {type(parsed)}")
            print(f"基本數(shù)據(jù)一致性: {parsed['string'] == self.complex_data['string']}")
            
            return True
            
        except Exception as e:
            print(f"JSON錯(cuò)誤: {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("自定義序列化結(jié)果:")
            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"自定義序列化錯(cuò)誤: {e}")
            return False
    
    def demonstrate_json_performance(self):
        """演示JSON性能考慮"""
        print("=== JSON性能優(yōu)化 ===")
        
        # 創(chuàng)建大型測試數(shù)據(jù)
        large_data = {
            'users': [{'id': i, 'name': f'user_{i}', 'data': list(range(100))} 
                     for i in range(1000)],
            'metadata': {'timestamp': datetime.now().isoformat()}
        }
        
        # 比較不同選項(xiàng)的性能
        options = [
            ('默認(rèn)', {}),
            ('無縮進(jìn)', {'indent': None}),
            ('分隔符優(yōu)化', {'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} 字節(jié), 序列化: {serialize_time:.4f}s, 反序列化: {deserialize_time:.4f}s, 總計(jì): {total_time:.4f}s")
                
            except Exception as e:
                print(f"{name} 錯(cuò)誤: {e}")
    
    def demonstrate_json_schema(self):
        """演示JSON Schema驗(yàn)證"""
        print("=== JSON Schema驗(yàn)證 ===")
        
        # 簡單的schema驗(yàn)證示例
        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"]
        }
        
        # 測試數(shù)據(jù)
        test_cases = [
            {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'},  # 有效
            {'name': 'Bob', 'age': -5},  # 年齡無效
            {'age': 25},  # 缺少必填字段
            {'name': 123, 'age': 25}  # 名稱類型錯(cuò)誤
        ]
        
        # 簡單驗(yàn)證函數(shù)（實(shí)際應(yīng)用中應(yīng)該使用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} 應(yīng)該是字符串類型")
                    elif expected_type == 'number' and not isinstance(value, (int, float)):
                        errors.append(f"字段 {field} 應(yīng)該是數(shù)字類型")
                    elif expected_type == 'array' and not isinstance(value, list):
                        errors.append(f"字段 {field} 應(yīng)該是數(shù)組類型")
            
            return len(errors) == 0, errors
        
        # 測試驗(yàn)證
        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):
        """注冊內(nèi)置類型處理器"""
        # 基本類型
        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"序列化錯(cuò)誤: {e}")
    
    def deserialize(self, serialized_data):
        """反序列化數(shù)據(jù)"""
        if not isinstance(serialized_data, (list, tuple)) or len(serialized_data) != 2:
            raise ValueError("無效的序列化數(shù)據(jù)格式")
        
        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']
            
            # 動(dòng)態(tài)導(dǎo)入模塊（生產(chǎn)環(huán)境需要更安全的方法）
            try:
                module = __import__(module_name, fromlist=[class_name])
                obj_class = getattr(module, class_name)
                
                # 創(chuàng)建對象實(shí)例
                instance = obj_class.__new__(obj_class)
                
                # 恢復(fù)屬性
                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"反序列化對象錯(cuò)誤: {e}")
        else:
            raise ValueError(f"未知的類型標(biāo)識: {type_name}")
    
    def to_json_compatible(self, obj):
        """轉(zhuǎn)換為JSON兼容格式"""
        serialized = self.serialize(obj)
        return serialized
    
    def from_json_compatible(self, data):
        """從JSON兼容格式恢復(fù)"""
        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})"
    
    # 創(chuàng)建框架實(shí)例
    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)
            }
        }))
    
    # 測試數(shù)據(jù)
    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"序列化結(jié)果: {serialized}")
        
        # 轉(zhuǎn)換為JSON兼容格式
        json_compatible = framework.to_json_compatible(nested_data)
        print(f"JSON兼容格式: {json_compatible}")
        
        # 反序列化
        deserialized = framework.deserialize(serialized)
        print(f"反序列化成功: {type(deserialized)}")
        
        # 驗(yàn)證
        print(f"數(shù)據(jù)一致性: {nested_data['string'] == deserialized['string']}")
        print(f"對象一致性: {nested_data['object'] == deserialized['object']}")
        
    except Exception as e:
        print(f"自定義序列化錯(cuò)誤: {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):
        """創(chuàng)建測試數(shù)據(jù)"""
        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):
        """性能基準(zhǔn)測試"""
        print("=== 序列化性能基準(zhǔn)測試 ===")
        
        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:
                # 預(yù)熱
                if name != 'pickle':  # pickle不需要預(yù)熱
                    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)
                
                # 驗(yàn)證數(shù)據(jù)一致性
                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)
                })
        
        # 顯示結(jié)果
        print(f"{'庫名':<12} {'序列化':<8} {'反序列化':<8} {'總計(jì)':<8} {'大小':<8} {'驗(yàn)證'}")
        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} 錯(cuò)誤: {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)} 字節(jié)")
            print(f"JSON大小: {len(json.dumps(self.test_data))} 字節(jié)")
            print(f"壓縮比: {len(json.dumps(self.test_data)) / len(packed):.2f}x")
            
            # 反序列化
            unpacked = msgpack.unpackb(packed)
            print(f"反序列化成功: {type(unpacked)}")
            print(f"數(shù)據(jù)一致性: {unpacked['users'][0]['name'] == self.test_data['users'][0]['name']}")
            
            # 顯示部分二進(jìn)制數(shù)據(jù)
            print(f"二進(jìn)制數(shù)據(jù) (前50字節(jié)): {packed[:50].hex(' ')}...")
            
        except Exception as e:
            print(f"MessagePack錯(cuò)誤: {e}")
    
    def demonstrate_orjson_features(self):
        """演示orjson特性"""
        print("=== orjson特性演示 ===")
        
        try:
            # orjson支持更多數(shù)據(jù)類型
            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)} 字節(jié)")
            
            # 反序列化
            deserialized = orjson.loads(serialized)
            print(f"反序列化成功: {type(deserialized)}")
            
            # orjson特性
            print("orjson支持:")
            print("  - 內(nèi)置datetime支持")
            print("  - 高性能C實(shí)現(xiàn)")
            print("  - 無GIL限制")
            print("  - 內(nèi)存效率高")
            
        except Exception as e:
            print(f"orjson錯(cuò)誤: {e}")

# 使用示例
def demo_performance_serialization():
    """性能序列化演示"""
    perf = HighPerformanceSerializer()
    
    # 運(yùn)行性能測試
    results = perf.benchmark_serializers()
    print()
    
    # 演示特定庫特性
    perf.demonstrate_msgpack_features()
    print()
    perf.demonstrate_orjson_features()

demo_performance_serialization()

class SecureSerialization:
    """
    安全序列化實(shí)踐
    """
    
    def demonstrate_pickle_security_risks(self):
        """演示pickle安全風(fēng)險(xiǎn)"""
        print("=== pickle安全風(fēng)險(xiǎn)演示 ===")
        
        # 危險(xiǎn)的pickle數(shù)據(jù)
        malicious_code = """
import os
os.system('echo "危險(xiǎn)操作被執(zhí)行"')
"""
        
        # 創(chuàng)建惡意pickle數(shù)據(jù)
        class MaliciousPayload:
            def __reduce__(self):
                return (eval, (malicious_code,))
        
        try:
            # 序列化惡意載荷
            malicious_pickle = pickle.dumps(MaliciousPayload())
            print(f"惡意pickle數(shù)據(jù)創(chuàng)建: {len(malicious_pickle)} 字節(jié)")
            
            # 警告：不要在實(shí)際環(huán)境中執(zhí)行以下代碼
            print("警告: 以下操作可能危險(xiǎn)，僅在受控環(huán)境中演示")
            
            # 演示反序列化風(fēng)險(xiǎn)（注釋掉實(shí)際執(zhí)行）
            # result = pickle.loads(malicious_pickle)
            # print(f"惡意代碼執(zhí)行結(jié)果: {result}")
            
            print("? 實(shí)際執(zhí)行已禁用，僅用于演示風(fēng)險(xiǎn)")
            
        except Exception as e:
            print(f"惡意代碼演示錯(cuò)誤: {e}")
    
    def demonstrate_safe_alternatives(self):
        """演示安全替代方案"""
        print("=== 安全序列化替代方案 ===")
        
        # 1. 使用JSON進(jìn)行安全數(shù)據(jù)交換
        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結(jié)果: {safe_result}")
        except Exception as e:
            print(f"安全unpickler錯(cuò)誤: {e}")
        
        # 3. 數(shù)據(jù)驗(yàn)證和清洗
        def sanitize_data(data):
            """數(shù)據(jù)清洗函數(shù)"""
            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"不安全的數(shù)據(jù)類型: {type(data)}")
        
        try:
            cleaned_data = sanitize_data(safe_data)
            print(f"數(shù)據(jù)清洗成功: {cleaned_data}")
        except Exception as e:
            print(f"數(shù)據(jù)清洗錯(cuò)誤: {e}")
    
    def demonstrate_encrypted_serialization(self):
        """演示加密序列化"""
        print("=== 加密序列化演示 ===")
        
        from cryptography.fernet import Fernet
        
        # 生成加密密鑰
        key = Fernet.generate_key()
        cipher = Fernet(key)
        
        # 要加密的數(shù)據(jù)
        sensitive_data = {
            'username': 'admin',
            'password': 'secret123',  # 實(shí)際應(yīng)用中應(yīng)該使用哈希
            'token': 'abcdef123456'
        }
        
        try:
            # 序列化后加密
            serialized = pickle.dumps(sensitive_data)
            encrypted = cipher.encrypt(serialized)
            
            print(f"原始數(shù)據(jù)大小: {len(serialized)} 字節(jié)")
            print(f"加密數(shù)據(jù)大小: {len(encrypted)} 字節(jié)")
            print(f"加密數(shù)據(jù): {encrypted[:30]}...")
            
            # 解密和反序列化
            decrypted = cipher.decrypt(encrypted)
            deserialized = pickle.loads(decrypted)
            
            print(f"解密成功: {deserialized['username']}")
            print(f"數(shù)據(jù)完整性: {deserialized == sensitive_data}")
            
        except Exception as e:
            print(f"加密序列化錯(cuò)誤: {e}")
    
    def demonstrate_signing_data(self):
        """演示數(shù)據(jù)簽名"""
        print("=== 數(shù)據(jù)簽名演示 ===")
        
        import hmac
        import hashlib
        
        # 共享密鑰（實(shí)際應(yīng)用中應(yīng)該安全存儲）
        secret_key = b'my_secret_key'
        
        data_to_sign = {'important': 'data', 'timestamp': time.time()}
        serialized_data = json.dumps(data_to_sign).encode('utf-8')
        
        # 創(chuàng)建簽名
        signature = hmac.new(secret_key, serialized_data, hashlib.sha256).hexdigest()
        
        # 組合數(shù)據(jù)和簽名
        signed_package = {
            'data': data_to_sign,
            'signature': signature
        }
        
        print(f"簽名數(shù)據(jù)包: {signed_package}")
        
        # 驗(yàn)證簽名
        def verify_signature(data, received_signature):
            """驗(yàn)證數(shù)據(jù)簽名"""
            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)
        
        # 測試驗(yàn)證
        is_valid = verify_signature(data_to_sign, signature)
        print(f"簽名驗(yàn)證: {'成功' 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:
    """
    基于序列化的配置管理系統(tǒng)
    """
    
    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"配置文件不存在，使用默認(rèn)配置: {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"配置文件加載錯(cuò)誤: {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"配置文件保存錯(cuò)誤: {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):
        """
        設(shè)置配置值
        """
        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):
        """
        驗(yàn)證配置有效性
        """
        # 簡單的驗(yàn)證邏輯
        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"缺少必填配置項(xiàng): {key}")
        
        # 驗(yàn)證端口范圍
        db_port = self.get_config_value('database.port')
        if db_port and not (0 < db_port < 65536):
            errors.append(f"數(shù)據(jù)庫端口無效: {db_port}")
        
        # 驗(yàn)證日志級別
        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("配置驗(yàn)證失敗:")
            for error in errors:
                print(f"  - {error}")
            return False
        else:
            print("配置驗(yàn)證成功")
            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("=== 配置管理系統(tǒng) ===")
    
    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')
    
    # 驗(yàn)證配置
    manager.validate_configuration()
    
    # 保存配置
    manager.save_configuration()
    
    # 讀取配置值
    app_name = manager.get_config_value('app.name')
    db_port = manager.get_config_value('database.port')
    print(f"應(yīng)用名稱: {app_name}")
    print(f"數(shù)據(jù)庫端口: {db_port}")
    
    # 清理
    if os.path.exists('demo_config.json'):
        os.remove('demo_config.json')

demo_configuration_manager()

class DistributedTaskQueue:
    """
    基于序列化的分布式任務(wù)隊(duì)列
    """
    
    def __init__(self, queue_name='default', serializer='pickle'):
        self.queue_name = queue_name
        self.serializer = serializer
        self.tasks = []
        
        # 創(chuàng)建序列化器
        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):
        """
        將任務(wù)加入隊(duì)列
        """
        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"任務(wù)已加入隊(duì)列: {task_id}")
        return task_id
    
    def serialize_queue(self, filename=None):
        """
        序列化任務(wù)隊(duì)列
        """
        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"隊(duì)列已序列化到文件: {filename}")
        
        return serialized_data
    
    def deserialize_queue(self, data=None, filename=None):
        """
        反序列化任務(wù)隊(duì)列
        """
        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"隊(duì)列已從{'文件' if filename else '數(shù)據(jù)'}加載: {len(self.tasks)} 個(gè)任務(wù)")
            return True
        
        return False
    
    def process_tasks(self, max_tasks=None):
        """
        處理任務(wù)
        """
        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"處理任務(wù) {i+1}: {task['id']}")
                try:
                    # 模擬任務(wù)處理
                    result = f"處理結(jié)果: {task['function']}({len(task['args'])} 參數(shù))"
                    task['status'] = 'completed'
                    task['result'] = result
                    task['completed_at'] = time.time()
                    processed += 1
                    
                    print(f"任務(wù)完成: {result}")
                    
                except Exception as e:
                    task['status'] = 'failed'
                    task['error'] = str(e)
                    print(f"任務(wù)失敗: {e}")
        
        return processed
    
    def get_queue_stats(self):
        """
        獲取隊(duì)列統(tǒng)計(jì)
        """
        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("隊(duì)列統(tǒng)計(jì):")
        for key, value in stats.items():
            print(f"  {key}: {value}")
        
        return stats
    
    def clear_queue(self, status=None):
        """
        清理隊(duì)列
        """
        if status:
            self.tasks = [t for t in self.tasks if t['status'] != status]
            print(f"已清理狀態(tài)為 {status} 的任務(wù)")
        else:
            self.tasks.clear()
            print("隊(duì)列已清空")

# 使用示例
def demo_task_queue():
    """任務(wù)隊(duì)列演示"""
    print("=== 分布式任務(wù)隊(duì)列 ===")
    
    # 創(chuàng)建任務(wù)函數(shù)
    def process_data(data, multiplier=1):
        return f"處理了 {len(data)} 條數(shù)據(jù)，結(jié)果: {[x * multiplier for x in data]}"
    
    def send_email(to, subject, body):
        return f"發(fā)送郵件到 {to}: {subject}"
    
    def generate_report(format='pdf', pages=10):
        return f"生成 {format} 報(bào)告，{pages} 頁"
    
    # 創(chuàng)建隊(duì)列
    queue = DistributedTaskQueue(serializer='json')
    
    # 添加任務(wù)
    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)
    
    # 顯示統(tǒng)計(jì)
    queue.get_queue_stats()
    
    # 序列化隊(duì)列
    serialized = queue.serialize_queue('task_queue.json')
    print(f"序列化數(shù)據(jù)大小: {len(serialized)} 字節(jié)")
    
    # 處理任務(wù)
    processed = queue.process_tasks(2)
    print(f"處理了 {processed} 個(gè)任務(wù)")
    
    # 更新統(tǒng)計(jì)
    queue.get_queue_stats()
    
    # 清理
    queue.clear_queue('completed')
    queue.get_queue_stats()
    
    # 保存最終狀態(tài)
    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()