第204集Kafka十万级TPS高性能存储架构实战：分区策略、批量处理、零拷贝的企业级优化方案

// 智能分区管理器
@Component
@Slf4j
public class IntelligentPartitionManager {
    
    private final KafkaAdminClient adminClient;
    private final ProducerMetrics producerMetrics;
    private final PartitionLoadBalancer loadBalancer;
    private final MeterRegistry meterRegistry;
    
    public IntelligentPartitionManager(KafkaAdminClient adminClient,
                                     ProducerMetrics producerMetrics,
                                     PartitionLoadBalancer loadBalancer,
                                     MeterRegistry meterRegistry) {
        this.adminClient = adminClient;
        this.producerMetrics = producerMetrics;
        this.loadBalancer = loadBalancer;
        this.meterRegistry = meterRegistry;
    }
    
    // 动态分区分配
    public void dynamicPartitionAllocation(String topic, int targetPartitions) {
        log.info("Starting dynamic partition allocation for topic: {}, target partitions: {}", 
                topic, targetPartitions);
        
        try {
            // 分析当前分区负载
            PartitionLoadAnalysis analysis = analyzePartitionLoad(topic);
            
            // 计算最优分区数量
            int optimalPartitions = calculateOptimalPartitions(analysis, targetPartitions);
            
            // 执行分区扩容
            if (optimalPartitions > analysis.getCurrentPartitions()) {
                expandPartitions(topic, optimalPartitions);
            }
            
            // 重新平衡分区负载
            rebalancePartitionLoad(topic);
            
            meterRegistry.counter("partition.allocation.completed").increment();
            log.info("Dynamic partition allocation completed for topic: {}", topic);
            
        } catch (Exception e) {
            log.error("Error in dynamic partition allocation", e);
            meterRegistry.counter("partition.allocation.error").increment();
        }
    }
    
    // 分析分区负载
    private PartitionLoadAnalysis analyzePartitionLoad(String topic) {
        PartitionLoadAnalysis analysis = new PartitionLoadAnalysis();
        
        try {
            // 获取分区元数据
            TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(topic))
                    .values().get(topic).get();
            
            List<PartitionInfo> partitions = topicDescription.partitions();
            analysis.setCurrentPartitions(partitions.size());
            
            // 分析每个分区的负载
            Map<Integer, PartitionLoad> partitionLoads = new HashMap<>();
            
            for (PartitionInfo partition : partitions) {
                PartitionLoad load = analyzeSinglePartitionLoad(topic, partition.partition());
                partitionLoads.put(partition.partition(), load);
            }
            
            analysis.setPartitionLoads(partitionLoads);
            
            // 计算负载统计
            calculateLoadStatistics(analysis);
            
        } catch (Exception e) {
            log.error("Error analyzing partition load", e);
            analysis.setError(e.getMessage());
        }
        
        return analysis;
    }
    
    // 分析单个分区负载
    private PartitionLoad analyzeSinglePartitionLoad(String topic, int partition) {
        PartitionLoad load = new PartitionLoad();
        
        try {
            // 获取分区消息数量
            long messageCount = getPartitionMessageCount(topic, partition);
            load.setMessageCount(messageCount);
            
            // 获取分区大小
            long partitionSize = getPartitionSize(topic, partition);
            load.setPartitionSize(partitionSize);
            
            // 获取分区写入速率
            double writeRate = getPartitionWriteRate(topic, partition);
            load.setWriteRate(writeRate);
            
            // 获取分区读取速率
            double readRate = getPartitionReadRate(topic, partition);
            load.setReadRate(readRate);
            
            // 计算负载分数
            double loadScore = calculateLoadScore(load);
            load.setLoadScore(loadScore);
            
        } catch (Exception e) {
            log.error("Error analyzing single partition load", e);
            load.setError(e.getMessage());
        }
        
        return load;
    }
    
    // 计算负载分数
    private double calculateLoadScore(PartitionLoad load) {
        // 综合考虑消息数量、分区大小、读写速率
        double messageScore = Math.log(load.getMessageCount() + 1) / 10.0;
        double sizeScore = Math.log(load.getPartitionSize() + 1) / 20.0;
        double rateScore = (load.getWriteRate() + load.getReadRate()) / 1000.0;
        
        return messageScore + sizeScore + rateScore;
    }
    
    // 计算负载统计
    private void calculateLoadStatistics(PartitionLoadAnalysis analysis) {
        Map<Integer, PartitionLoad> loads = analysis.getPartitionLoads();
        
        if (loads.isEmpty()) {
            return;
        }
        
        // 计算平均负载
        double avgLoad = loads.values().stream()
                .mapToDouble(PartitionLoad::getLoadScore)
                .average()
                .orElse(0.0);
        analysis.setAverageLoad(avgLoad);
        
        // 计算最大负载
        double maxLoad = loads.values().stream()
                .mapToDouble(PartitionLoad::getLoadScore)
                .max()
                .orElse(0.0);
        analysis.setMaxLoad(maxLoad);
        
        // 计算负载方差
        double variance = loads.values().stream()
                .mapToDouble(load -> Math.pow(load.getLoadScore() - avgLoad, 2))
                .average()
                .orElse(0.0);
        analysis.setLoadVariance(variance);
        
        // 计算负载不均衡度
        double imbalanceRatio = maxLoad / Math.max(avgLoad, 0.001);
        analysis.setImbalanceRatio(imbalanceRatio);
    }
    
    // 计算最优分区数量
    private int calculateOptimalPartitions(PartitionLoadAnalysis analysis, int targetPartitions) {
        // 基于负载不均衡度计算最优分区数
        double imbalanceRatio = analysis.getImbalanceRatio();
        
        if (imbalanceRatio > 2.0) {
            // 负载严重不均衡，需要增加分区
            return Math.min(analysis.getCurrentPartitions() * 2, targetPartitions);
        } else if (imbalanceRatio > 1.5) {
            // 负载轻微不均衡，适度增加分区
            return Math.min((int) (analysis.getCurrentPartitions() * 1.5), targetPartitions);
        } else {
            // 负载相对均衡，保持当前分区数
            return analysis.getCurrentPartitions();
        }
    }
    
    // 扩展分区
    private void expandPartitions(String topic, int newPartitionCount) {
        log.info("Expanding partitions for topic: {} to {}", topic, newPartitionCount);
        
        try {
            // 创建分区扩展请求
            Map<String, NewPartitions> partitionsToAdd = new HashMap<>();
            partitionsToAdd.put(topic, NewPartitions.increaseTo(newPartitionCount));
            
            // 执行分区扩展
            CreatePartitionsResult result = adminClient.createPartitions(partitionsToAdd);
            result.all().get(30, TimeUnit.SECONDS);
            
            meterRegistry.counter("partition.expansion.completed").increment();
            log.info("Partition expansion completed for topic: {}", topic);
            
        } catch (Exception e) {
            log.error("Error expanding partitions", e);
            meterRegistry.counter("partition.expansion.error").increment();
            throw new RuntimeException("Failed to expand partitions", e);
        }
    }
    
    // 重新平衡分区负载
    private void rebalancePartitionLoad(String topic) {
        log.info("Rebalancing partition load for topic: {}", topic);
        
        try {
            // 获取当前分区负载
            PartitionLoadAnalysis analysis = analyzePartitionLoad(topic);
            
            // 识别高负载分区
            List<Integer> highLoadPartitions = identifyHighLoadPartitions(analysis);
            
            // 识别低负载分区
            List<Integer> lowLoadPartitions = identifyLowLoadPartitions(analysis);
            
            // 执行负载重平衡
            executeLoadRebalance(topic, highLoadPartitions, lowLoadPartitions);
            
            meterRegistry.counter("partition.rebalance.completed").increment();
            log.info("Partition load rebalancing completed for topic: {}", topic);
            
        } catch (Exception e) {
            log.error("Error rebalancing partition load", e);
            meterRegistry.counter("partition.rebalance.error").increment();
        }
    }
    
    // 识别高负载分区
    private List<Integer> identifyHighLoadPartitions(PartitionLoadAnalysis analysis) {
        double threshold = analysis.getAverageLoad() * 1.5;
        
        return analysis.getPartitionLoads().entrySet().stream()
                .filter(entry -> entry.getValue().getLoadScore() > threshold)
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
    }
    
    // 识别低负载分区
    private List<Integer> identifyLowLoadPartitions(PartitionLoadAnalysis analysis) {
        double threshold = analysis.getAverageLoad() * 0.5;
        
        return analysis.getPartitionLoads().entrySet().stream()
                .filter(entry -> entry.getValue().getLoadScore() < threshold)
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
    }
    
    // 执行负载重平衡
    private void executeLoadRebalance(String topic, List<Integer> highLoadPartitions, 
                                    List<Integer> lowLoadPartitions) {
        // 这里可以实现负载重平衡逻辑
        // 比如调整分区副本分布、迁移数据等
        
        log.info("Executing load rebalance for topic: {}, high load: {}, low load: {}", 
                topic, highLoadPartitions, lowLoadPartitions);
    }
    
    // 获取分区消息数量
    private long getPartitionMessageCount(String topic, int partition) {
        try {
            // 这里可以从Kafka管理API获取分区消息数量
            return 1000000; // 模拟数据
        } catch (Exception e) {
            log.error("Error getting partition message count", e);
            return 0;
        }
    }
    
    // 获取分区大小
    private long getPartitionSize(String topic, int partition) {
        try {
            // 这里可以从文件系统获取分区大小
            return 1024 * 1024 * 1024; // 模拟1GB
        } catch (Exception e) {
            log.error("Error getting partition size", e);
            return 0;
        }
    }
    
    // 获取分区写入速率
    private double getPartitionWriteRate(String topic, int partition) {
        try {
            // 这里可以从监控系统获取写入速率
            return 1000.0; // 模拟1000 msg/s
        } catch (Exception e) {
            log.error("Error getting partition write rate", e);
            return 0.0;
        }
    }
    
    // 获取分区读取速率
    private double getPartitionReadRate(String topic, int partition) {
        try {
            // 这里可以从监控系统获取读取速率
            return 500.0; // 模拟500 msg/s
        } catch (Exception e) {
            log.error("Error getting partition read rate", e);
            return 0.0;
        }
    }
}

// 高性能分区路由器
@Component
@Slf4j
public class HighPerformancePartitionRouter {
    
    private final Map<String, PartitionRouter> topicRouters = new ConcurrentHashMap<>();
    private final PartitionLoadBalancer loadBalancer;
    private final MeterRegistry meterRegistry;
    
    public HighPerformancePartitionRouter(PartitionLoadBalancer loadBalancer,
                                        MeterRegistry meterRegistry) {
        this.loadBalancer = loadBalancer;
        this.meterRegistry = meterRegistry;
    }
    
    // 智能分区路由
    public int routeToPartition(String topic, String key, Object value, 
                               PartitioningStrategy strategy) {
        try {
            PartitionRouter router = getOrCreateRouter(topic);
            
            int partition = router.route(key, value, strategy);
            
            // 更新路由指标
            meterRegistry.counter("partition.routing.success")
                    .tag("topic", topic)
                    .tag("strategy", strategy.name())
                    .increment();
            
            return partition;
            
        } catch (Exception e) {
            log.error("Error routing to partition", e);
            meterRegistry.counter("partition.routing.error")
                    .tag("topic", topic)
                    .increment();
            
            // 返回默认分区
            return 0;
        }
    }
    
    // 获取或创建分区路由器
    private PartitionRouter getOrCreateRouter(String topic) {
        return topicRouters.computeIfAbsent(topic, t -> new PartitionRouter(t, loadBalancer));
    }
    
    // 分区路由器实现
    private static class PartitionRouter {
        private final String topic;
        private final PartitionLoadBalancer loadBalancer;
        private final AtomicInteger roundRobinCounter = new AtomicInteger(0);
        
        public PartitionRouter(String topic, PartitionLoadBalancer loadBalancer) {
            this.topic = topic;
            this.loadBalancer = loadBalancer;
        }
        
        public int route(String key, Object value, PartitioningStrategy strategy) {
            switch (strategy) {
                case KEY_HASH:
                    return routeByKeyHash(key);
                case ROUND_ROBIN:
                    return routeByRoundRobin();
                case LOAD_BALANCED:
                    return routeByLoadBalance();
                case CUSTOM:
                    return routeByCustom(key, value);
                default:
                    return routeByKeyHash(key);
            }
        }
        
        // 基于键哈希的路由
        private int routeByKeyHash(String key) {
            if (key == null) {
                return 0;
            }
            
            int partitionCount = loadBalancer.getPartitionCount(topic);
            return Math.abs(key.hashCode()) % partitionCount;
        }
        
        // 轮询路由
        private int routeByRoundRobin() {
            int partitionCount = loadBalancer.getPartitionCount(topic);
            return roundRobinCounter.getAndIncrement() % partitionCount;
        }
        
        // 负载均衡路由
        private int routeByLoadBalance() {
            return loadBalancer.getLeastLoadedPartition(topic);
        }
        
        // 自定义路由
        private int routeByCustom(String key, Object value) {
            // 这里可以实现自定义路由逻辑
            return routeByKeyHash(key);
        }
    }
    
    // 分区策略枚举
    public enum PartitioningStrategy {
        KEY_HASH("键哈希"),
        ROUND_ROBIN("轮询"),
        LOAD_BALANCED("负载均衡"),
        CUSTOM("自定义");
        
        private final String description;
        
        PartitioningStrategy(String description) {
            this.description = description;
        }
        
        public String getDescription() {
            return description;
        }
    }
}

// 高性能批量处理器
@Component
@Slf4j
public class HighPerformanceBatchProcessor {
    
    private final Map<String, BatchBuffer> topicBuffers = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);
    private final MeterRegistry meterRegistry;
    
    // 批量处理配置
    private final BatchProcessingConfig config;
    
    public HighPerformanceBatchProcessor(MeterRegistry meterRegistry,
                                        BatchProcessingConfig config) {
        this.meterRegistry = meterRegistry;
        this.config = config;
        
        // 启动批量处理任务
        startBatchProcessingTasks();
    }
    
    // 添加消息到批量缓冲区
    public void addToBatch(String topic, ProducerRecord<String, Object> record) {
        try {
            BatchBuffer buffer = getOrCreateBuffer(topic);
            buffer.addRecord(record);
            
            // 检查是否需要立即发送
            if (shouldFlushImmediately(buffer)) {
                flushBuffer(topic);
            }
            
        } catch (Exception e) {
            log.error("Error adding message to batch", e);
            meterRegistry.counter("batch.add.error").increment();
        }
    }
    
    // 获取或创建批量缓冲区
    private BatchBuffer getOrCreateBuffer(String topic) {
        return topicBuffers.computeIfAbsent(topic, t -> new BatchBuffer(t, config));
    }
    
    // 检查是否需要立即刷新
    private boolean shouldFlushImmediately(BatchBuffer buffer) {
        return buffer.getRecordCount() >= config.getMaxBatchSize() ||
               buffer.getSize() >= config.getMaxBatchSizeBytes() ||
               buffer.getAge() >= config.getMaxBatchDelayMs();
    }
    
    // 刷新缓冲区
    public void flushBuffer(String topic) {
        try {
            BatchBuffer buffer = topicBuffers.get(topic);
            if (buffer != null && !buffer.isEmpty()) {
                List<ProducerRecord<String, Object>> records = buffer.drainRecords();
                
                if (!records.isEmpty()) {
                    // 发送批量消息
                    sendBatchMessages(topic, records);
                    
                    // 更新指标
                    meterRegistry.counter("batch.flush.success")
                            .tag("topic", topic)
                            .increment();
                    meterRegistry.gauge("batch.size", records.size())
                            .tag("topic", topic)
                            .register();
                }
            }
            
        } catch (Exception e) {
            log.error("Error flushing buffer for topic: {}", topic, e);
            meterRegistry.counter("batch.flush.error")
                    .tag("topic", topic)
                    .increment();
        }
    }
    
    // 发送批量消息
    private void sendBatchMessages(String topic, List<ProducerRecord<String, Object>> records) {
        try {
            // 这里可以实现批量发送逻辑
            // 比如使用KafkaProducer的send方法
            
            log.debug("Sending batch of {} messages for topic: {}", records.size(), topic);
            
            // 模拟批量发送
            Thread.sleep(1); // 模拟网络延迟
            
        } catch (Exception e) {
            log.error("Error sending batch messages", e);
            throw new RuntimeException("Failed to send batch messages", e);
        }
    }
    
    // 启动批量处理任务
    private void startBatchProcessingTasks() {
        // 定时刷新任务
        scheduler.scheduleAtFixedRate(() -> {
            try {
                flushAllBuffers();
            } catch (Exception e) {
                log.error("Error in scheduled batch flush", e);
            }
        }, config.getFlushIntervalMs(), config.getFlushIntervalMs(), TimeUnit.MILLISECONDS);
        
        // 监控任务
        scheduler.scheduleAtFixedRate(() -> {
            try {
                updateBatchMetrics();
            } catch (Exception e) {
                log.error("Error updating batch metrics", e);
            }
        }, 0, 5, TimeUnit.SECONDS);
    }
    
    // 刷新所有缓冲区
    private void flushAllBuffers() {
        for (String topic : topicBuffers.keySet()) {
            try {
                flushBuffer(topic);
            } catch (Exception e) {
                log.error("Error flushing buffer for topic: {}", topic, e);
            }
        }
    }
    
    // 更新批量指标
    private void updateBatchMetrics() {
        for (Map.Entry<String, BatchBuffer> entry : topicBuffers.entrySet()) {
            String topic = entry.getKey();
            BatchBuffer buffer = entry.getValue();
            
            meterRegistry.gauge("batch.buffer.size", buffer.getRecordCount())
                    .tag("topic", topic)
                    .register();
            meterRegistry.gauge("batch.buffer.age", buffer.getAge())
                    .tag("topic", topic)
                    .register();
        }
    }
    
    // 批量缓冲区实现
    private static class BatchBuffer {
        private final String topic;
        private final BatchProcessingConfig config;
        private final List<ProducerRecord<String, Object>> records = new ArrayList<>();
        private final long createdTime = System.currentTimeMillis();
        private final AtomicInteger size = new AtomicInteger(0);
        
        public BatchBuffer(String topic, BatchProcessingConfig config) {
            this.topic = topic;
            this.config = config;
        }
        
        public synchronized void addRecord(ProducerRecord<String, Object> record) {
            records.add(record);
            size.addAndGet(estimateRecordSize(record));
        }
        
        public synchronized List<ProducerRecord<String, Object>> drainRecords() {
            List<ProducerRecord<String, Object>> result = new ArrayList<>(records);
            records.clear();
            size.set(0);
            return result;
        }
        
        public synchronized boolean isEmpty() {
            return records.isEmpty();
        }
        
        public int getRecordCount() {
            return records.size();
        }
        
        public int getSize() {
            return size.get();
        }
        
        public long getAge() {
            return System.currentTimeMillis() - createdTime;
        }
        
        private int estimateRecordSize(ProducerRecord<String, Object> record) {
            int size = 0;
            
            if (record.key() != null) {
                size += record.key().length();
            }
            
            if (record.value() != null) {
                size += record.value().toString().length();
            }
            
            return size;
        }
    }
}

// 批量处理配置
@ConfigurationProperties(prefix = "kafka.batch")
@Data
public class BatchProcessingConfig {
    
    // 最大批量大小
    private int maxBatchSize = 1000;
    
    // 最大批量大小（字节）
    private int maxBatchSizeBytes = 1024 * 1024; // 1MB
    
    // 最大批量延迟（毫秒）
    private long maxBatchDelayMs = 100;
    
    // 刷新间隔（毫秒）
    private long flushIntervalMs = 50;
    
    // 压缩类型
    private String compressionType = "snappy";
    
    // 是否启用压缩
    private boolean compressionEnabled = true;
    
    // 批量重试次数
    private int batchRetryCount = 3;
    
    // 批量重试延迟（毫秒）
    private long batchRetryDelayMs = 100;
    
    // 是否启用批量确认
    private boolean batchAckEnabled = true;
    
    // 批量确认超时（毫秒）
    private long batchAckTimeoutMs = 5000;
}

// 零拷贝传输管理器
@Component
@Slf4j
public class ZeroCopyTransferManager {
    
    private final FileChannelManager fileChannelManager;
    private final MemoryMappedFileManager memoryMappedManager;
    private final SendFileManager sendFileManager;
    private final MeterRegistry meterRegistry;
    
    public ZeroCopyTransferManager(FileChannelManager fileChannelManager,
                                  MemoryMappedFileManager memoryMappedManager,
                                  SendFileManager sendFileManager,
                                  MeterRegistry meterRegistry) {
        this.fileChannelManager = fileChannelManager;
        this.memoryMappedManager = memoryMappedManager;
        this.sendFileManager = sendFileManager;
        this.meterRegistry = meterRegistry;
    }
    
    // 零拷贝文件传输
    public void transferFileZeroCopy(String sourcePath, String targetPath) {
        log.info("Starting zero-copy file transfer: {} -> {}", sourcePath, targetPath);
        
        try {
            long startTime = System.currentTimeMillis();
            
            // 使用sendfile进行零拷贝传输
            long bytesTransferred = sendFileManager.transferFile(sourcePath, targetPath);
            
            long duration = System.currentTimeMillis() - startTime;
            
            // 更新指标
            meterRegistry.counter("zero_copy.transfer.success").increment();
            meterRegistry.gauge("zero_copy.transfer.bytes", bytesTransferred).register();
            meterRegistry.gauge("zero_copy.transfer.duration", duration).register();
            
            log.info("Zero-copy file transfer completed: {} bytes in {} ms", 
                    bytesTransferred, duration);
            
        } catch (Exception e) {
            log.error("Error in zero-copy file transfer", e);
            meterRegistry.counter("zero_copy.transfer.error").increment();
            throw new RuntimeException("Failed to transfer file with zero-copy", e);
        }
    }
    
    // 零拷贝内存映射
    public MappedByteBuffer mapFileZeroCopy(String filePath, long position, long size) {
        log.info("Mapping file with zero-copy: {}, position: {}, size: {}", 
                filePath, position, size);
        
        try {
            MappedByteBuffer buffer = memoryMappedManager.mapFile(filePath, position, size);
            
            meterRegistry.counter("zero_copy.memory_map.success").increment();
            meterRegistry.gauge("zero_copy.memory_map.size", size).register();
            
            return buffer;
            
        } catch (Exception e) {
            log.error("Error in zero-copy memory mapping", e);
            meterRegistry.counter("zero_copy.memory_map.error").increment();
            throw new RuntimeException("Failed to map file with zero-copy", e);
        }
    }
    
    // 零拷贝网络传输
    public void transferNetworkZeroCopy(SocketChannel channel, FileChannel fileChannel, 
                                       long position, long count) {
        log.info("Starting zero-copy network transfer: position: {}, count: {}", 
                position, count);
        
        try {
            long startTime = System.currentTimeMillis();
            
            // 使用transferTo进行零拷贝传输
            long bytesTransferred = fileChannel.transferTo(position, count, channel);
            
            long duration = System.currentTimeMillis() - startTime;
            
            // 更新指标
            meterRegistry.counter("zero_copy.network_transfer.success").increment();
            meterRegistry.gauge("zero_copy.network_transfer.bytes", bytesTransferred).register();
            meterRegistry.gauge("zero_copy.network_transfer.duration", duration).register();
            
            log.info("Zero-copy network transfer completed: {} bytes in {} ms", 
                    bytesTransferred, duration);
            
        } catch (Exception e) {
            log.error("Error in zero-copy network transfer", e);
            meterRegistry.counter("zero_copy.network_transfer.error").increment();
            throw new RuntimeException("Failed to transfer network data with zero-copy", e);
        }
    }
    
    // 批量零拷贝传输
    public void batchTransferZeroCopy(List<TransferTask> tasks) {
        log.info("Starting batch zero-copy transfer: {} tasks", tasks.size());
        
        try {
            long totalBytesTransferred = 0;
            long startTime = System.currentTimeMillis();
            
            for (TransferTask task : tasks) {
                try {
                    long bytesTransferred = executeTransferTask(task);
                    totalBytesTransferred += bytesTransferred;
                    
                } catch (Exception e) {
                    log.error("Error in transfer task: {}", task, e);
                }
            }
            
            long duration = System.currentTimeMillis() - startTime;
            
            // 更新指标
            meterRegistry.counter("zero_copy.batch_transfer.success").increment();
            meterRegistry.gauge("zero_copy.batch_transfer.total_bytes", totalBytesTransferred).register();
            meterRegistry.gauge("zero_copy.batch_transfer.duration", duration).register();
            
            log.info("Batch zero-copy transfer completed: {} bytes in {} ms", 
                    totalBytesTransferred, duration);
            
        } catch (Exception e) {
            log.error("Error in batch zero-copy transfer", e);
            meterRegistry.counter("zero_copy.batch_transfer.error").increment();
        }
    }
    
    // 执行传输任务
    private long executeTransferTask(TransferTask task) throws IOException {
        switch (task.getTransferType()) {
            case FILE_TO_FILE:
                return sendFileManager.transferFile(task.getSourcePath(), task.getTargetPath());
            case FILE_TO_NETWORK:
                return transferFileToNetwork(task);
            case MEMORY_TO_NETWORK:
                return transferMemoryToNetwork(task);
            default:
                throw new IllegalArgumentException("Unsupported transfer type: " + task.getTransferType());
        }
    }
    
    // 文件到网络传输
    private long transferFileToNetwork(TransferTask task) throws IOException {
        try (FileChannel fileChannel = FileChannel.open(Paths.get(task.getSourcePath()), StandardOpenOption.READ);
             SocketChannel socketChannel = SocketChannel.open()) {
            
            socketChannel.connect(new InetSocketAddress(task.getTargetHost(), task.getTargetPort()));
            
            return fileChannel.transferTo(task.getPosition(), task.getCount(), socketChannel);
        }
    }
    
    // 内存到网络传输
    private long transferMemoryToNetwork(TransferTask task) throws IOException {
        try (SocketChannel socketChannel = SocketChannel.open()) {
            socketChannel.connect(new InetSocketAddress(task.getTargetHost(), task.getTargetPort()));
            
            ByteBuffer buffer = task.getBuffer();
            long bytesTransferred = 0;
            
            while (buffer.hasRemaining()) {
                bytesTransferred += socketChannel.write(buffer);
            }
            
            return bytesTransferred;
        }
    }
}

// SendFile管理器
@Component
@Slf4j
public class SendFileManager {
    
    private final MeterRegistry meterRegistry;
    
    public SendFileManager(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    
    // 使用sendfile传输文件
    public long transferFile(String sourcePath, String targetPath) throws IOException {
        try (FileChannel sourceChannel = FileChannel.open(Paths.get(sourcePath), StandardOpenOption.READ);
             FileChannel targetChannel = FileChannel.open(Paths.get(targetPath), 
                     StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
            
            long fileSize = sourceChannel.size();
            long bytesTransferred = 0;
            
            while (bytesTransferred < fileSize) {
                long transferred = sourceChannel.transferTo(bytesTransferred, 
                        fileSize - bytesTransferred, targetChannel);
                
                if (transferred == 0) {
                    break; // 没有更多数据可传输
                }
                
                bytesTransferred += transferred;
            }
            
            meterRegistry.gauge("sendfile.transfer.bytes", bytesTransferred).register();
            return bytesTransferred;
            
        } catch (IOException e) {
            log.error("Error in sendfile transfer", e);
            meterRegistry.counter("sendfile.transfer.error").increment();
            throw e;
        }
    }
    
    // 使用sendfile传输到网络
    public long transferFileToNetwork(String filePath, SocketChannel socketChannel) throws IOException {
        try (FileChannel fileChannel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ)) {
            
            long fileSize = fileChannel.size();
            long bytesTransferred = 0;
            
            while (bytesTransferred < fileSize) {
                long transferred = fileChannel.transferTo(bytesTransferred, 
                        fileSize - bytesTransferred, socketChannel);
                
                if (transferred == 0) {
                    break; // 没有更多数据可传输
                }
                
                bytesTransferred += transferred;
            }
            
            meterRegistry.gauge("sendfile.network_transfer.bytes", bytesTransferred).register();
            return bytesTransferred;
            
        } catch (IOException e) {
            log.error("Error in sendfile network transfer", e);
            meterRegistry.counter("sendfile.network_transfer.error").increment();
            throw e;
        }
    }
    
    // 批量sendfile传输
    public void batchTransferFiles(List<FileTransferTask> tasks) {
        log.info("Starting batch sendfile transfer: {} tasks", tasks.size());
        
        try {
            long totalBytesTransferred = 0;
            long startTime = System.currentTimeMillis();
            
            for (FileTransferTask task : tasks) {
                try {
                    long bytesTransferred = transferFile(task.getSourcePath(), task.getTargetPath());
                    totalBytesTransferred += bytesTransferred;
                    
                } catch (Exception e) {
                    log.error("Error in file transfer task: {}", task, e);
                }
            }
            
            long duration = System.currentTimeMillis() - startTime;
            
            meterRegistry.counter("sendfile.batch_transfer.success").increment();
            meterRegistry.gauge("sendfile.batch_transfer.total_bytes", totalBytesTransferred).register();
            meterRegistry.gauge("sendfile.batch_transfer.duration", duration).register();
            
            log.info("Batch sendfile transfer completed: {} bytes in {} ms", 
                    totalBytesTransferred, duration);
            
        } catch (Exception e) {
            log.error("Error in batch sendfile transfer", e);
            meterRegistry.counter("sendfile.batch_transfer.error").increment();
        }
    }
}

// 顺序写入管理器
@Component
@Slf4j
public class SequentialWriteManager {
    
    private final Map<String, SequentialWriter> topicWriters = new ConcurrentHashMap<>();
    private final FileChannelManager fileChannelManager;
    private final MeterRegistry meterRegistry;
    
    public SequentialWriteManager(FileChannelManager fileChannelManager,
                                MeterRegistry meterRegistry) {
        this.fileChannelManager = fileChannelManager;
        this.meterRegistry = meterRegistry;
    }
    
    // 顺序写入消息
    public void writeSequentially(String topic, int partition, byte[] data) {
        try {
            SequentialWriter writer = getOrCreateWriter(topic, partition);
            writer.write(data);
            
            meterRegistry.counter("sequential_write.success")
                    .tag("topic", topic)
                    .tag("partition", String.valueOf(partition))
                    .increment();
            
        } catch (Exception e) {
            log.error("Error in sequential write", e);
            meterRegistry.counter("sequential_write.error")
                    .tag("topic", topic)
                    .tag("partition", String.valueOf(partition))
                    .increment();
        }
    }
    
    // 批量顺序写入
    public void writeBatchSequentially(String topic, int partition, List<byte[]> dataList) {
        try {
            SequentialWriter writer = getOrCreateWriter(topic, partition);
            writer.writeBatch(dataList);
            
            meterRegistry.counter("sequential_write.batch.success")
                    .tag("topic", topic)
                    .tag("partition", String.valueOf(partition))
                    .increment();
            meterRegistry.gauge("sequential_write.batch.size", dataList.size())
                    .tag("topic", topic)
                    .tag("partition", String.valueOf(partition))
                    .register();
            
        } catch (Exception e) {
            log.error("Error in batch sequential write", e);
            meterRegistry.counter("sequential_write.batch.error")
                    .tag("topic", topic)
                    .tag("partition", String.valueOf(partition))
                    .increment();
        }
    }
    
    // 获取或创建顺序写入器
    private SequentialWriter getOrCreateWriter(String topic, int partition) {
        String key = topic + "-" + partition;
        return topicWriters.computeIfAbsent(key, k -> new SequentialWriter(topic, partition, fileChannelManager));
    }
    
    // 顺序写入器实现
    private static class SequentialWriter {
        private final String topic;
        private final int partition;
        private final FileChannelManager fileChannelManager;
        private final String logFilePath;
        private FileChannel fileChannel;
        private long currentOffset = 0;
        
        public SequentialWriter(String topic, int partition, FileChannelManager fileChannelManager) {
            this.topic = topic;
            this.partition = partition;
            this.fileChannelManager = fileChannelManager;
            this.logFilePath = String.format("/data/kafka-logs/%s-%d/00000000000000000000.log", topic, partition);
            
            initializeWriter();
        }
        
        // 初始化写入器
        private void initializeWriter() {
            try {
                Path logPath = Paths.get(logFilePath);
                Files.createDirectories(logPath.getParent());
                
                if (!Files.exists(logPath)) {
                    Files.createFile(logPath);
                }
                
                fileChannel = FileChannel.open(logPath, 
                        StandardOpenOption.WRITE, 
                        StandardOpenOption.APPEND);
                
                currentOffset = fileChannel.size();
                
            } catch (IOException e) {
                throw new RuntimeException("Failed to initialize sequential writer", e);
            }
        }
        
        // 写入数据
        public synchronized void write(byte[] data) throws IOException {
            ByteBuffer buffer = ByteBuffer.wrap(data);
            
            while (buffer.hasRemaining()) {
                fileChannel.write(buffer);
            }
            
            currentOffset += data.length;
            
            // 强制刷新到磁盘
            fileChannel.force(false);
        }
        
        // 批量写入数据
        public synchronized void writeBatch(List<byte[]> dataList) throws IOException {
            for (byte[] data : dataList) {
                write(data);
            }
        }
        
        // 获取当前偏移量
        public long getCurrentOffset() {
            return currentOffset;
        }
        
        // 关闭写入器
        public void close() throws IOException {
            if (fileChannel != null) {
                fileChannel.close();
            }
        }
    }
}

// 高性能压缩管理器
@Component
@Slf4j
public class HighPerformanceCompressionManager {
    
    private final Map<CompressionType, Compressor> compressors = new HashMap<>();
    private final MeterRegistry meterRegistry;
    
    public HighPerformanceCompressionManager(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        initializeCompressors();
    }
    
    // 初始化压缩器
    private void initializeCompressors() {
        compressors.put(CompressionType.NONE, new NoCompressor());
        compressors.put(CompressionType.GZIP, new GzipCompressor());
        compressors.put(CompressionType.SNAPPY, new SnappyCompressor());
        compressors.put(CompressionType.LZ4, new LZ4Compressor());
        compressors.put(CompressionType.ZSTD, new ZstdCompressor());
        
        log.info("Compressors initialized: {}", compressors.keySet());
    }
    
    // 压缩数据
    public CompressedData compress(byte[] data, CompressionType type) {
        try {
            long startTime = System.currentTimeMillis();
            
            Compressor compressor = compressors.get(type);
            if (compressor == null) {
                throw new IllegalArgumentException("Unsupported compression type: " + type);
            }
            
            byte[] compressedData = compressor.compress(data);
            long duration = System.currentTimeMillis() - startTime;
            
            // 计算压缩率
            double compressionRatio = (double) compressedData.length / data.length;
            
            // 更新指标
            meterRegistry.counter("compression.success")
                    .tag("type", type.name())
                    .increment();
            meterRegistry.gauge("compression.ratio", compressionRatio)
                    .tag("type", type.name())
                    .register();
            meterRegistry.gauge("compression.duration", duration)
                    .tag("type", type.name())
                    .register();
            
            return CompressedData.builder()
                    .originalSize(data.length)
                    .compressedSize(compressedData.length)
                    .compressionRatio(compressionRatio)
                    .compressionType(type)
                    .data(compressedData)
                    .build();
            
        } catch (Exception e) {
            log.error("Error compressing data with type: {}", type, e);
            meterRegistry.counter("compression.error")
                    .tag("type", type.name())
                    .increment();
            throw new RuntimeException("Failed to compress data", e);
        }
    }
    
    // 解压缩数据
    public byte[] decompress(CompressedData compressedData) {
        try {
            long startTime = System.currentTimeMillis();
            
            Compressor compressor = compressors.get(compressedData.getCompressionType());
            if (compressor == null) {
                throw new IllegalArgumentException("Unsupported compression type: " + 
                        compressedData.getCompressionType());
            }
            
            byte[] decompressedData = compressor.decompress(compressedData.getData());
            long duration = System.currentTimeMillis() - startTime;
            
            // 更新指标
            meterRegistry.counter("decompression.success")
                    .tag("type", compressedData.getCompressionType().name())
                    .increment();
            meterRegistry.gauge("decompression.duration", duration)
                    .tag("type", compressedData.getCompressionType().name())
                    .register();
            
            return decompressedData;
            
        } catch (Exception e) {
            log.error("Error decompressing data", e);
            meterRegistry.counter("decompression.error")
                    .tag("type", compressedData.getCompressionType().name())
                    .increment();
            throw new RuntimeException("Failed to decompress data", e);
        }
    }
    
    // 批量压缩
    public List<CompressedData> batchCompress(List<byte[]> dataList, CompressionType type) {
        log.info("Starting batch compression: {} items with type: {}", dataList.size(), type);
        
        try {
            List<CompressedData> compressedList = new ArrayList<>();
            long startTime = System.currentTimeMillis();
            
            for (byte[] data : dataList) {
                CompressedData compressed = compress(data, type);
                compressedList.add(compressed);
            }
            
            long duration = System.currentTimeMillis() - startTime;
            
            // 更新指标
            meterRegistry.counter("compression.batch.success")
                    .tag("type", type.name())
                    .increment();
            meterRegistry.gauge("compression.batch.duration", duration)
                    .tag("type", type.name())
                    .register();
            meterRegistry.gauge("compression.batch.size", dataList.size())
                    .tag("type", type.name())
                    .register();
            
            log.info("Batch compression completed: {} items in {} ms", dataList.size(), duration);
            
            return compressedList;
            
        } catch (Exception e) {
            log.error("Error in batch compression", e);
            meterRegistry.counter("compression.batch.error")
                    .tag("type", type.name())
                    .increment();
            throw new RuntimeException("Failed to batch compress data", e);
        }
    }
    
    // 选择最优压缩算法
    public CompressionType selectOptimalCompression(byte[] data) {
        try {
            CompressionType bestType = CompressionType.NONE;
            double bestScore = Double.MAX_VALUE;
            
            for (CompressionType type : CompressionType.values()) {
                if (type == CompressionType.NONE) {
                    continue;
                }
                
                CompressedData compressed = compress(data, type);
                double score = calculateCompressionScore(compressed);
                
                if (score < bestScore) {
                    bestScore = score;
                    bestType = type;
                }
            }
            
            return bestType;
            
        } catch (Exception e) {
            log.error("Error selecting optimal compression", e);
            return CompressionType.NONE;
        }
    }
    
    // 计算压缩分数
    private double calculateCompressionScore(CompressedData compressed) {
        // 综合考虑压缩率和压缩时间
        double compressionRatio = compressed.getCompressionRatio();
        double timeScore = 1.0; // 这里可以根据压缩时间计算
        
        return compressionRatio + timeScore;
    }
}

// 压缩器接口
public interface Compressor {
    byte[] compress(byte[] data) throws IOException;
    byte[] decompress(byte[] compressedData) throws IOException;
}

// Snappy压缩器
@Component
public class SnappyCompressor implements Compressor {
    
    @Override
    public byte[] compress(byte[] data) throws IOException {
        return Snappy.compress(data);
    }
    
    @Override
    public byte[] decompress(byte[] compressedData) throws IOException {
        return Snappy.uncompress(compressedData);
    }
}

// LZ4压缩器
@Component
public class LZ4Compressor implements Compressor {
    
    private final LZ4Factory factory = LZ4Factory.fastestInstance();
    
    @Override
    public byte[] compress(byte[] data) throws IOException {
        LZ4Compressor compressor = factory.fastCompressor();
        int maxCompressedLength = compressor.maxCompressedLength(data.length);
        byte[] compressed = new byte[maxCompressedLength];
        
        int compressedLength = compressor.compress(data, 0, data.length, compressed, 0, maxCompressedLength);
        
        return Arrays.copyOf(compressed, compressedLength);
    }
    
    @Override
    public byte[] decompress(byte[] compressedData) throws IOException {
        LZ4FastDecompressor decompressor = factory.fastDecompressor();
        
        // 这里需要知道原始数据长度，实际实现中需要存储这个信息
        int originalLength = compressedData.length * 2; // 假设压缩率为50%
        byte[] decompressed = new byte[originalLength];
        
        int decompressedLength = decompressor.decompress(compressedData, decompressed);
        
        return Arrays.copyOf(decompressed, decompressedLength);
    }
}

// Zstd压缩器
@Component
public class ZstdCompressor implements Compressor {
    
    @Override
    public byte[] compress(byte[] data) throws IOException {
        return Zstd.compress(data);
    }
    
    @Override
    public byte[] decompress(byte[] compressedData) throws IOException {
        return Zstd.decompress(compressedData);
    }
}

// 内存映射文件管理器
@Component
@Slf4j
public class MemoryMappedFileManager {
    
    private final Map<String, MappedFile> mappedFiles = new ConcurrentHashMap<>();
    private final MeterRegistry meterRegistry;
    
    public MemoryMappedFileManager(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    
    // 映射文件到内存
    public MappedByteBuffer mapFile(String filePath, long position, long size) {
        log.info("Mapping file to memory: {}, position: {}, size: {}", filePath, position, size);
        
        try {
            Path path = Paths.get(filePath);
            
            if (!Files.exists(path)) {
                Files.createDirectories(path.getParent());
                Files.createFile(path);
            }
            
            try (FileChannel channel = FileChannel.open(path, 
                    StandardOpenOption.READ, 
                    StandardOpenOption.WRITE)) {
                
                MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_WRITE, position, size);
                
                // 记录映射信息
                MappedFile mappedFile = new MappedFile(filePath, position, size, buffer);
                mappedFiles.put(filePath, mappedFile);
                
                meterRegistry.counter("memory_map.success").increment();
                meterRegistry.gauge("memory_map.size", size).register();
                
                return buffer;
                
            }
            
        } catch (Exception e) {
            log.error("Error mapping file to memory: {}", filePath, e);
            meterRegistry.counter("memory_map.error").increment();
            throw new RuntimeException("Failed to map file to memory", e);
        }
    }
    
    // 取消内存映射
    public void unmapFile(String filePath) {
        log.info("Unmapping file from memory: {}", filePath);
        
        try {
            MappedFile mappedFile = mappedFiles.remove(filePath);
            if (mappedFile != null) {
                // 强制刷新到磁盘
                mappedFile.getBuffer().force();
                
                meterRegistry.counter("memory_unmap.success").increment();
                log.info("File unmapped successfully: {}", filePath);
            }
            
        } catch (Exception e) {
            log.error("Error unmapping file from memory: {}", filePath, e);
            meterRegistry.counter("memory_unmap.error").increment();
        }
    }
    
    // 批量映射文件
    public Map<String, MappedByteBuffer> batchMapFiles(List<FileMappingTask> tasks) {
        log.info("Starting batch file mapping: {} tasks", tasks.size());
        
        Map<String, MappedByteBuffer> mappedBuffers = new HashMap<>();
        
        try {
            long startTime = System.currentTimeMillis();
            
            for (FileMappingTask task : tasks) {
                try {
                    MappedByteBuffer buffer = mapFile(task.getFilePath(), 
                            task.getPosition(), task.getSize());
                    mappedBuffers.put(task.getFilePath(), buffer);
                    
                } catch (Exception e) {
                    log.error("Error mapping file: {}", task.getFilePath(), e);
                }
            }
            
            long duration = System.currentTimeMillis() - startTime;
            
            meterRegistry.counter("memory_map.batch.success").increment();
            meterRegistry.gauge("memory_map.batch.duration", duration).register();
            meterRegistry.gauge("memory_map.batch.size", tasks.size()).register();
            
            log.info("Batch file mapping completed: {} files in {} ms", tasks.size(), duration);
            
        } catch (Exception e) {
            log.error("Error in batch file mapping", e);
            meterRegistry.counter("memory_map.batch.error").increment();
        }
        
        return mappedBuffers;
    }
    
    // 获取映射文件信息
    public MappedFileInfo getMappedFileInfo(String filePath) {
        MappedFile mappedFile = mappedFiles.get(filePath);
        
        if (mappedFile != null) {
            return MappedFileInfo.builder()
                    .filePath(filePath)
                    .position(mappedFile.getPosition())
                    .size(mappedFile.getSize())
                    .mapped(true)
                    .build();
        } else {
            return MappedFileInfo.builder()
                    .filePath(filePath)
                    .mapped(false)
                    .build();
        }
    }
    
    // 获取所有映射文件信息
    public List<MappedFileInfo> getAllMappedFileInfo() {
        return mappedFiles.values().stream()
                .map(mappedFile -> MappedFileInfo.builder()
                        .filePath(mappedFile.getFilePath())
                        .position(mappedFile.getPosition())
                        .size(mappedFile.getSize())
                        .mapped(true)
                        .build())
                .collect(Collectors.toList());
    }
    
    // 清理所有映射
    public void cleanupAllMappings() {
        log.info("Cleaning up all memory mappings: {} files", mappedFiles.size());
        
        try {
            for (String filePath : new HashSet<>(mappedFiles.keySet())) {
                unmapFile(filePath);
            }
            
            meterRegistry.counter("memory_map.cleanup.success").increment();
            log.info("All memory mappings cleaned up successfully");
            
        } catch (Exception e) {
            log.error("Error cleaning up memory mappings", e);
            meterRegistry.counter("memory_map.cleanup.error").increment();
        }
    }
    
    // 映射文件信息类
    @Data
    @Builder
    private static class MappedFile {
        private final String filePath;
        private final long position;
        private final long size;
        private final MappedByteBuffer buffer;
    }
}

// Kafka性能监控配置
@Configuration
public class KafkaPerformanceMonitoringConfig {
    
    @Bean
    public AlertRule highLatencyAlertRule() {
        return AlertRule.builder()
                .name("Kafka High Latency")
                .description("Kafka message processing latency is too high")
                .condition("kafka.latency.p99 > 100")
                .severity(AlertSeverity.WARNING)
                .enabled(true)
                .build();
    }
    
    @Bean
    public AlertRule lowThroughputAlertRule() {
        return AlertRule.builder()
                .name("Kafka Low Throughput")
                .description("Kafka throughput is below expected level")
                .condition("kafka.throughput.tps < 50000")
                .severity(AlertSeverity.WARNING)
                .enabled(true)
                .build();
    }
    
    @Bean
    public AlertRule partitionImbalanceAlertRule() {
        return AlertRule.builder()
                .name("Kafka Partition Imbalance")
                .description("Kafka partition load is imbalanced")
                .condition("kafka.partition.imbalance_ratio > 2.0")
                .severity(AlertSeverity.WARNING)
                .enabled(true)
                .build();
    }
    
    @Bean
    public AlertRule compressionFailureAlertRule() {
        return AlertRule.builder()
                .name("Kafka Compression Failure")
                .description("Kafka compression is failing")
                .condition("kafka.compression.error > 10 per minute")
                .severity(AlertSeverity.CRITICAL)
                .enabled(true)
                .build();
    }
    
    @Bean
    public AlertRule zeroCopyFailureAlertRule() {
        return AlertRule.builder()
                .name("Kafka Zero-Copy Failure")
                .description("Kafka zero-copy transfer is failing")
                .condition("kafka.zero_copy.transfer.error > 5 per minute")
                .severity(AlertSeverity.CRITICAL)
                .enabled(true)
                .build();
    }
}