第186集大量序列化对象产生与性能优化

// 序列化框架性能对比
public class SerializationFrameworkComparison {

    // JSON序列化 (Jackson)
    public void jacksonSerialization(Object obj) {
        ObjectMapper mapper = new ObjectMapper();
        try {
            String json = mapper.writeValueAsString(obj);
            Object deserialized = mapper.readValue(json, obj.getClass());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // Protobuf序列化
    public void protobufSerialization(MessageLite message) {
        try {
            byte[] data = message.toByteArray();
            MessageLite deserialized = message.getDefaultInstanceForType()
                .getParserForType().parseFrom(data);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // Kryo序列化
    public void kryoSerialization(Object obj) {
        Kryo kryo = new Kryo();
        Output output = new Output(1024);
        kryo.writeObject(output, obj);
        byte[] data = output.toBytes();

        Input input = new Input(data);
        Object deserialized = kryo.readObject(input, obj.getClass());
    }

    // Avro序列化
    public void avroSerialization(GenericRecord record) {
        try {
            DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(record.getSchema());
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
            writer.write(record, encoder);
            encoder.flush();
            byte[] data = out.toByteArray();

            DatumReader<GenericRecord> reader = new GenericDatumReader<>(record.getSchema());
            BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null);
            GenericRecord deserialized = reader.read(null, decoder);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

// 序列化性能测试
@Component
public class SerializationPerformanceTest {
    private final ObjectMapper jacksonMapper = new ObjectMapper();
    private final Kryo kryo = new Kryo();

    public void performanceTest() {
        TestObject testObj = createTestObject();
        int iterations = 100000;

        // Jackson性能测试
        long jacksonTime = testJackson(testObj, iterations);
        System.out.println("Jackson序列化时间: " + jacksonTime + "ms");

        // Kryo性能测试
        long kryoTime = testKryo(testObj, iterations);
        System.out.println("Kryo序列化时间: " + kryoTime + "ms");

        // Protobuf性能测试
        long protobufTime = testProtobuf(testObj, iterations);
        System.out.println("Protobuf序列化时间: " + protobufTime + "ms");
    }

    private long testJackson(Object obj, int iterations) {
        long startTime = System.currentTimeMillis();

        for (int i = 0; i < iterations; i++) {
            try {
                String json = jacksonMapper.writeValueAsString(obj);
                Object deserialized = jacksonMapper.readValue(json, obj.getClass());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        return System.currentTimeMillis() - startTime;
    }

    private long testKryo(Object obj, int iterations) {
        long startTime = System.currentTimeMillis();

        for (int i = 0; i < iterations; i++) {
            Output output = new Output(1024);
            kryo.writeObject(output, obj);
            byte[] data = output.toBytes();

            Input input = new Input(data);
            Object deserialized = kryo.readObject(input, obj.getClass());
        }

        return System.currentTimeMillis() - startTime;
    }

    private TestObject createTestObject() {
        TestObject obj = new TestObject();
        obj.setId(1L);
        obj.setName("Test Object");
        obj.setValue(123.45);
        obj.setTimestamp(System.currentTimeMillis());
        return obj;
    }
}

// 序列化对象池实现
@Component
public class SerializationObjectPool {
    private final Queue<Output> outputPool = new ConcurrentLinkedQueue<>();
    private final Queue<Input> inputPool = new ConcurrentLinkedQueue<>();
    private final Queue<ByteArrayOutputStream> byteArrayPool = new ConcurrentLinkedQueue<>();

    private final int maxPoolSize = 100;
    private final int initialBufferSize = 1024;

    public Output borrowOutput() {
        Output output = outputPool.poll();
        if (output == null) {
            output = new Output(initialBufferSize);
        }
        return output;
    }

    public void returnOutput(Output output) {
        if (outputPool.size() < maxPoolSize) {
            output.clear();
            outputPool.offer(output);
        }
    }

    public Input borrowInput(byte[] data) {
        Input input = inputPool.poll();
        if (input == null) {
            input = new Input(data);
        } else {
            input.setBuffer(data);
        }
        return input;
    }

    public void returnInput(Input input) {
        if (inputPool.size() < maxPoolSize) {
            inputPool.offer(input);
        }
    }

    public ByteArrayOutputStream borrowByteArrayOutputStream() {
        ByteArrayOutputStream out = byteArrayPool.poll();
        if (out == null) {
            out = new ByteArrayOutputStream(initialBufferSize);
        } else {
            out.reset();
        }
        return out;
    }

    public void returnByteArrayOutputStream(ByteArrayOutputStream out) {
        if (byteArrayPool.size() < maxPoolSize) {
            byteArrayPool.offer(out);
        }
    }
}

// 线程本地对象池
@Component
public class ThreadLocalObjectPool {
    private final ThreadLocal<Output> threadLocalOutput = new ThreadLocal<Output>() {
        @Override
        protected Output initialValue() {
            return new Output(1024);
        }
    };

    private final ThreadLocal<Input> threadLocalInput = new ThreadLocal<Input>() {
        @Override
        protected Input initialValue() {
            return new Input(1024);
        }
    };

    private final ThreadLocal<ByteArrayOutputStream> threadLocalByteArray = 
        new ThreadLocal<ByteArrayOutputStream>() {
            @Override
            protected ByteArrayOutputStream initialValue() {
                return new ByteArrayOutputStream(1024);
            }
        };

    public Output getOutput() {
        Output output = threadLocalOutput.get();
        output.clear();
        return output;
    }

    public Input getInput(byte[] data) {
        Input input = threadLocalInput.get();
        input.setBuffer(data);
        return input;
    }

    public ByteArrayOutputStream getByteArrayOutputStream() {
        ByteArrayOutputStream out = threadLocalByteArray.get();
        out.reset();
        return out;
    }

    public void clear() {
        threadLocalOutput.remove();
        threadLocalInput.remove();
        threadLocalByteArray.remove();
    }
}

// 对象复用策略
@Service
public class ObjectReuseStrategy {
    private final SerializationObjectPool objectPool;
    private final ThreadLocalObjectPool threadLocalPool;

    public ObjectReuseStrategy(SerializationObjectPool objectPool,
                              ThreadLocalObjectPool threadLocalPool) {
        this.objectPool = objectPool;
        this.threadLocalPool = threadLocalPool;
    }

    public byte[] serializeWithReuse(Object obj, Kryo kryo) {
        Output output = threadLocalPool.getOutput();
        try {
            kryo.writeObject(output, obj);
            return output.toBytes();
        } finally {
            // 不需要归还，因为是ThreadLocal
        }
    }

    public Object deserializeWithReuse(byte[] data, Kryo kryo, Class<?> clazz) {
        Input input = threadLocalPool.getInput(data);
        try {
            return kryo.readObject(input, clazz);
        } finally {
            // 不需要归还，因为是ThreadLocal
        }
    }

    public byte[] serializeWithPool(Object obj, Kryo kryo) {
        Output output = objectPool.borrowOutput();
        try {
            kryo.writeObject(output, obj);
            return output.toBytes();
        } finally {
            objectPool.returnOutput(output);
        }
    }

    public Object deserializeWithPool(byte[] data, Kryo kryo, Class<?> clazz) {
        Input input = objectPool.borrowInput(data);
        try {
            return kryo.readObject(input, clazz);
        } finally {
            objectPool.returnInput(input);
        }
    }
}

// 内存池管理
@Component
public class MemoryPoolManager {
    private final Queue<byte[]> smallBufferPool = new ConcurrentLinkedQueue<>();
    private final Queue<byte[]> mediumBufferPool = new ConcurrentLinkedQueue<>();
    private final Queue<byte[]> largeBufferPool = new ConcurrentLinkedQueue<>();

    private final int smallBufferSize = 1024;
    private final int mediumBufferSize = 4096;
    private final int largeBufferSize = 16384;

    public byte[] borrowBuffer(int size) {
        if (size <= smallBufferSize) {
            return borrowFromPool(smallBufferPool, smallBufferSize);
        } else if (size <= mediumBufferSize) {
            return borrowFromPool(mediumBufferPool, mediumBufferSize);
        } else if (size <= largeBufferSize) {
            return borrowFromPool(largeBufferPool, largeBufferSize);
        } else {
            return new byte[size];
        }
    }

    public void returnBuffer(byte[] buffer) {
        if (buffer.length == smallBufferSize) {
            smallBufferPool.offer(buffer);
        } else if (buffer.length == mediumBufferSize) {
            mediumBufferPool.offer(buffer);
        } else if (buffer.length == largeBufferSize) {
            largeBufferPool.offer(buffer);
        }
        // 其他大小的buffer不回收
    }

    private byte[] borrowFromPool(Queue<byte[]> pool, int size) {
        byte[] buffer = pool.poll();
        if (buffer == null) {
            buffer = new byte[size];
        }
        return buffer;
    }
}

// 内存监控与预警
@Component
public class MemoryMonitor {
    private final MemoryMXBean memoryBean;
    private final List<GarbageCollectorMXBean> gcBeans;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public MemoryMonitor() {
        this.memoryBean = ManagementFactory.getMemoryMXBean();
        this.gcBeans = ManagementFactory.getGarbageCollectorMXBeans();

        // 启动内存监控
        scheduler.scheduleAtFixedRate(this::monitorMemory, 0, 5, TimeUnit.SECONDS);
    }

    private void monitorMemory() {
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
        double usageRatio = (double) heapUsage.getUsed() / heapUsage.getMax();

        if (usageRatio > 0.8) {
            System.out.println("警告: 内存使用率过高: " + String.format("%.2f%%", usageRatio * 100));

            // 触发内存清理
            System.gc();

            // 记录GC信息
            for (GarbageCollectorMXBean gcBean : gcBeans) {
                System.out.println("GC: " + gcBean.getName() + 
                                 ", 次数: " + gcBean.getCollectionCount() + 
                                 ", 时间: " + gcBean.getCollectionTime() + "ms");
            }
        }
    }

    public boolean isMemoryHigh() {
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
        double usageRatio = (double) heapUsage.getUsed() / heapUsage.getMax();
        return usageRatio > 0.8;
    }
}

// 内存泄漏防护
@Component
public class MemoryLeakPrevention {
    private final WeakHashMap<String, Object> cache = new WeakHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    @PostConstruct
    public void init() {
        // 定期清理缓存
        scheduler.scheduleAtFixedRate(this::cleanupCache, 0, 10, TimeUnit.MINUTES);

        // 定期检查内存使用
        scheduler.scheduleAtFixedRate(this::checkMemoryUsage, 0, 1, TimeUnit.MINUTES);
    }

    private void cleanupCache() {
        synchronized (cache) {
            cache.clear();
        }
        System.gc();
    }

    private void checkMemoryUsage() {
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();

        double usageRatio = (double) heapUsage.getUsed() / heapUsage.getMax();
        if (usageRatio > 0.9) {
            System.out.println("内存使用率过高: " + String.format("%.2f%%", usageRatio * 100));
            cleanupCache();
        }
    }

    public void addToCache(String key, Object value) {
        synchronized (cache) {
            cache.put(key, value);
        }
    }

    public Object getFromCache(String key) {
        synchronized (cache) {
            return cache.get(key);
        }
    }
}

// 序列化结果缓存
@Service
public class SerializationCacheService {
    private final Cache<String, byte[]> serializationCache;
    private final Cache<String, Class<?>> classCache;

    public SerializationCacheService() {
        this.serializationCache = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(30, TimeUnit.MINUTES)
            .build();

        this.classCache = Caffeine.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build();
    }

    public byte[] getSerializedData(String key) {
        return serializationCache.getIfPresent(key);
    }

    public void putSerializedData(String key, byte[] data) {
        serializationCache.put(key, data);
    }

    public Class<?> getClass(String className) {
        return classCache.get(className, name -> {
            try {
                return Class.forName(name);
            } catch (ClassNotFoundException e) {
                throw new RuntimeException("Class not found: " + name, e);
            }
        });
    }

    public void clearCache() {
        serializationCache.invalidateAll();
        classCache.invalidateAll();
    }
}

// 对象模板缓存
@Service
public class ObjectTemplateCache {
    private final Cache<String, Object> templateCache;
    private final Kryo kryo;

    public ObjectTemplateCache() {
        this.templateCache = Caffeine.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build();

        this.kryo = new Kryo();
        kryo.setRegistrationRequired(false);
    }

    public Object getTemplate(String className) {
        return templateCache.get(className, name -> {
            try {
                Class<?> clazz = Class.forName(name);
                return kryo.newInstance(clazz);
            } catch (Exception e) {
                throw new RuntimeException("Failed to create template for: " + name, e);
            }
        });
    }

    public void putTemplate(String className, Object template) {
        templateCache.put(className, template);
    }

    public void clearCache() {
        templateCache.invalidateAll();
    }
}

// 并发序列化处理
@Service
public class ConcurrentSerializationService {
    private final ExecutorService executorService;
    private final SerializationObjectPool objectPool;
    private final Kryo kryo;

    public ConcurrentSerializationService(SerializationObjectPool objectPool) {
        this.objectPool = objectPool;
        this.executorService = Executors.newFixedThreadPool(10);
        this.kryo = new Kryo();
        kryo.setRegistrationRequired(false);
    }

    public CompletableFuture<byte[]> serializeAsync(Object obj) {
        return CompletableFuture.supplyAsync(() -> {
            Output output = objectPool.borrowOutput();
            try {
                kryo.writeObject(output, obj);
                return output.toBytes();
            } finally {
                objectPool.returnOutput(output);
            }
        }, executorService);
    }

    public CompletableFuture<Object> deserializeAsync(byte[] data, Class<?> clazz) {
        return CompletableFuture.supplyAsync(() -> {
            Input input = objectPool.borrowInput(data);
            try {
                return kryo.readObject(input, clazz);
            } finally {
                objectPool.returnInput(input);
            }
        }, executorService);
    }

    public CompletableFuture<List<byte[]>> batchSerializeAsync(List<Object> objects) {
        List<CompletableFuture<byte[]>> futures = objects.stream()
            .map(this::serializeAsync)
            .collect(Collectors.toList());

        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList()));
    }
}

// 批量处理优化
@Service
public class BatchProcessingService {
    private final SerializationObjectPool objectPool;
    private final Kryo kryo;

    public BatchProcessingService(SerializationObjectPool objectPool) {
        this.objectPool = objectPool;
        this.kryo = new Kryo();
        kryo.setRegistrationRequired(false);
    }

    public List<byte[]> batchSerialize(List<Object> objects) {
        List<byte[]> results = new ArrayList<>(objects.size());

        for (Object obj : objects) {
            Output output = objectPool.borrowOutput();
            try {
                kryo.writeObject(output, obj);
                results.add(output.toBytes());
            } finally {
                objectPool.returnOutput(output);
            }
        }

        return results;
    }

    public List<Object> batchDeserialize(List<byte[]> dataList, Class<?> clazz) {
        List<Object> results = new ArrayList<>(dataList.size());

        for (byte[] data : dataList) {
            Input input = objectPool.borrowInput(data);
            try {
                Object obj = kryo.readObject(input, clazz);
                results.add(obj);
            } finally {
                objectPool.returnInput(input);
            }
        }

        return results;
    }

    public void processBatch(List<Object> objects, Consumer<Object> processor) {
        int batchSize = 100;

        for (int i = 0; i < objects.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, objects.size());
            List<Object> batch = objects.subList(i, endIndex);

            // 处理批次
            batch.forEach(processor);

            // 清理引用
            batch.clear();

            // 强制GC
            if (i % (batchSize * 10) == 0) {
                System.gc();
            }
        }
    }
}

// 序列化性能监控
@Component
public class SerializationPerformanceMonitor {
    private final MeterRegistry meterRegistry;
    private final Timer serializationTimer;
    private final Timer deserializationTimer;
    private final Counter serializationCounter;
    private final Counter deserializationCounter;

    public SerializationPerformanceMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.serializationTimer = Timer.builder("serialization.time")
            .register(meterRegistry);
        this.deserializationTimer = Timer.builder("deserialization.time")
            .register(meterRegistry);
        this.serializationCounter = Counter.builder("serialization.count")
            .register(meterRegistry);
        this.deserializationCounter = Counter.builder("deserialization.count")
            .register(meterRegistry);
    }

    public <T> T monitorSerialization(Supplier<T> operation, String operationName) {
        Timer.Sample sample = Timer.start(meterRegistry);
        try {
            T result = operation.get();
            serializationCounter.increment();
            return result;
        } finally {
            sample.stop(Timer.builder("serialization.time")
                .tag("operation", operationName)
                .register(meterRegistry));
        }
    }

    public <T> T monitorDeserialization(Supplier<T> operation, String operationName) {
        Timer.Sample sample = Timer.start(meterRegistry);
        try {
            T result = operation.get();
            deserializationCounter.increment();
            return result;
        } finally {
            sample.stop(Timer.builder("deserialization.time")
                .tag("operation", operationName)
                .register(meterRegistry));
        }
    }
}

// 性能调优建议
@Service
public class PerformanceTuningService {
    private final MemoryMonitor memoryMonitor;
    private final SerializationPerformanceMonitor performanceMonitor;

    public PerformanceTuningService(MemoryMonitor memoryMonitor,
                                   SerializationPerformanceMonitor performanceMonitor) {
        this.memoryMonitor = memoryMonitor;
        this.performanceMonitor = performanceMonitor;
    }

    public void optimizePerformance() {
        if (memoryMonitor.isMemoryHigh()) {
            System.out.println("内存使用率过高，开始性能优化...");

            // 1. 清理缓存
            System.gc();

            // 2. 调整对象池大小
            adjustObjectPoolSize();

            // 3. 优化序列化参数
            optimizeSerializationParameters();
        }
    }

    private void adjustObjectPoolSize() {
        // 根据内存使用情况调整对象池大小
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();

        long usedMemory = heapUsage.getUsed();
        long maxMemory = heapUsage.getMax();

        if ((double) usedMemory / maxMemory > 0.8) {
            System.out.println("建议减少对象池大小");
        } else if ((double) usedMemory / maxMemory < 0.5) {
            System.out.println("可以增加对象池大小");
        }
    }

    private void optimizeSerializationParameters() {
        // 优化序列化参数
        System.out.println("优化序列化参数...");

        // 可以在这里添加具体的优化逻辑
    }
}

// 微服务通信优化
@Service
public class MicroserviceCommunicationService {
    private final SerializationObjectPool objectPool;
    private final Kryo kryo;
    private final RestTemplate restTemplate;

    public MicroserviceCommunicationService(SerializationObjectPool objectPool,
                                           RestTemplate restTemplate) {
        this.objectPool = objectPool;
        this.restTemplate = restTemplate;
        this.kryo = new Kryo();
        kryo.setRegistrationRequired(false);
    }

    public <T> T callService(String url, Object request, Class<T> responseType) {
        // 序列化请求
        byte[] requestData = serializeWithPool(request);

        // 发送请求
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
        HttpEntity<byte[]> entity = new HttpEntity<>(requestData, headers);

        ResponseEntity<byte[]> response = restTemplate.postForEntity(url, entity, byte[].class);

        // 反序列化响应
        return deserializeWithPool(response.getBody(), responseType);
    }

    private byte[] serializeWithPool(Object obj) {
        Output output = objectPool.borrowOutput();
        try {
            kryo.writeObject(output, obj);
            return output.toBytes();
        } finally {
            objectPool.returnOutput(output);
        }
    }

    private <T> T deserializeWithPool(byte[] data, Class<T> clazz) {
        Input input = objectPool.borrowInput(data);
        try {
            return kryo.readObject(input, clazz);
        } finally {
            objectPool.returnInput(input);
        }
    }
}

// 缓存系统优化
@Service
public class CacheSystemOptimization {
    private final RedisTemplate<String, Object> redisTemplate;
    private final SerializationObjectPool objectPool;
    private final Kryo kryo;

    public CacheSystemOptimization(RedisTemplate<String, Object> redisTemplate,
                                   SerializationObjectPool objectPool) {
        this.redisTemplate = redisTemplate;
        this.objectPool = objectPool;
        this.kryo = new Kryo();
        kryo.setRegistrationRequired(false);
    }

    public void putToCache(String key, Object value) {
        byte[] serializedData = serializeWithPool(value);
        redisTemplate.opsForValue().set(key, serializedData);
    }

    public <T> T getFromCache(String key, Class<T> clazz) {
        byte[] data = (byte[]) redisTemplate.opsForValue().get(key);
        if (data != null) {
            return deserializeWithPool(data, clazz);
        }
        return null;
    }

    public void batchPutToCache(Map<String, Object> keyValueMap) {
        Map<String, byte[]> serializedMap = new HashMap<>();

        for (Map.Entry<String, Object> entry : keyValueMap.entrySet()) {
            byte[] serializedData = serializeWithPool(entry.getValue());
            serializedMap.put(entry.getKey(), serializedData);
        }

        redisTemplate.opsForValue().multiSet(serializedMap);
    }

    private byte[] serializeWithPool(Object obj) {
        Output output = objectPool.borrowOutput();
        try {
            kryo.writeObject(output, obj);
            return output.toBytes();
        } finally {
            objectPool.returnOutput(output);
        }
    }

    private <T> T deserializeWithPool(byte[] data, Class<T> clazz) {
        Input input = objectPool.borrowInput(data);
        try {
            return kryo.readObject(input, clazz);
        } finally {
            objectPool.returnInput(input);
        }
    }
}