第189集Kafka集群架构设计与高可用部署

// Topic分区策略
@Service
public class TopicPartitionStrategy {
    private final AdminClient adminClient;
    
    public TopicPartitionStrategy(AdminClient adminClient) {
        this.adminClient = adminClient;
    }
    
    // 创建Topic
    public void createTopic(String topicName, int partitions, int replicationFactor) {
        try {
            NewTopic newTopic = new NewTopic(topicName, partitions, (short) replicationFactor);
            
            CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
            result.all().get(30, TimeUnit.SECONDS);
            
            System.out.println("Topic " + topicName + " 创建成功");
        } catch (Exception e) {
            throw new RuntimeException("创建Topic失败", e);
        }
    }
    
    // 增加分区
    public void increasePartitions(String topicName, int newPartitionCount) {
        try {
            Map<String, NewPartitions> partitions = new HashMap<>();
            partitions.put(topicName, NewPartitions.increaseTo(newPartitionCount));
            
            CreatePartitionsResult result = adminClient.createPartitions(partitions);
            result.all().get(30, TimeUnit.SECONDS);
            
            System.out.println("Topic " + topicName + " 分区数增加到 " + newPartitionCount);
        } catch (Exception e) {
            throw new RuntimeException("增加分区失败", e);
        }
    }
    
    // 获取Topic信息
    public TopicDescription getTopicInfo(String topicName) {
        try {
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topicName));
            return result.topicNameValues().get(topicName).get(30, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new RuntimeException("获取Topic信息失败", e);
        }
    }
    
    // 计算分区数量
    public int calculatePartitionCount(int expectedThroughput, int partitionThroughput) {
        return Math.max(1, (expectedThroughput + partitionThroughput - 1) / partitionThroughput);
    }
    
    // 选择分区策略
    public int selectPartition(String key, int partitionCount) {
        if (key == null) {
            return ThreadLocalRandom.current().nextInt(partitionCount);
        }
        return Math.abs(key.hashCode()) % partitionCount;
    }
}

// 生产者优化配置
@Service
public class KafkaProducerOptimization {
    private final KafkaProducer<String, String> producer;
    
    public KafkaProducerOptimization() {
        Properties props = new Properties();
        
        // 基础配置
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // 性能优化配置
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 16KB
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 5ms
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
        
        // 内存优化
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32MB
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000); // 60s
        
        // 网络优化
        props.put(ProducerConfig.SEND_BUFFER_CONFIG, 131072); // 128KB
        props.put(ProducerConfig.RECEIVE_BUFFER_CONFIG, 32768); // 32KB
        
        this.producer = new KafkaProducer<>(props);
    }
    
    // 异步发送
    public void sendAsync(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                System.err.println("发送失败: " + exception.getMessage());
            } else {
                System.out.println("发送成功: " + metadata.topic() + "-" + metadata.partition() + "-" + metadata.offset());
            }
        });
    }
    
    // 批量发送
    public void sendBatch(String topic, List<ProducerRecord<String, String>> records) {
        for (ProducerRecord<String, String> record : records) {
            producer.send(record);
        }
        producer.flush();
    }
    
    // 事务发送
    public void sendTransactional(String topic, String key, String value) {
        producer.beginTransaction();
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            producer.send(record);
            producer.commitTransaction();
        } catch (Exception e) {
            producer.abortTransaction();
            throw new RuntimeException("事务发送失败", e);
        }
    }
}

// 消费者优化配置
@Service
public class KafkaConsumerOptimization {
    private final KafkaConsumer<String, String> consumer;
    
    public KafkaConsumerOptimization() {
        Properties props = new Properties();
        
        // 基础配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "optimized-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        
        // 性能优化配置
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576); // 1MB
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        
        // 提交策略
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        
        // 偏移量配置
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        
        this.consumer = new KafkaConsumer<>(props);
    }
    
    // 批量消费
    public void consumeBatch(String topic, Consumer<List<ConsumerRecord<String, String>>> processor) {
        consumer.subscribe(Collections.singletonList(topic));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            if (!records.isEmpty()) {
                List<ConsumerRecord<String, String>> recordList = new ArrayList<>();
                for (ConsumerRecord<String, String> record : records) {
                    recordList.add(record);
                }
                
                processor.accept(recordList);
                
                // 手动提交偏移量
                consumer.commitSync();
            }
        }
    }
    
    // 流式消费
    public void consumeStream(String topic, Consumer<ConsumerRecord<String, String>> processor) {
        consumer.subscribe(Collections.singletonList(topic));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            for (ConsumerRecord<String, String> record : records) {
                processor.accept(record);
            }
            
            // 批量提交偏移量
            consumer.commitSync();
        }
    }
    
    // 指定分区消费
    public void consumePartition(String topic, int partition, Consumer<ConsumerRecord<String, String>> processor) {
        TopicPartition topicPartition = new TopicPartition(topic, partition);
        consumer.assign(Collections.singletonList(topicPartition));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            for (ConsumerRecord<String, String> record : records) {
                processor.accept(record);
            }
            
            consumer.commitSync();
        }
    }
}

// 集群监控服务
@Service
public class KafkaClusterMonitor {
    private final AdminClient adminClient;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public KafkaClusterMonitor(AdminClient adminClient) {
        this.adminClient = adminClient;
        
        // 启动监控
        scheduler.scheduleAtFixedRate(this::monitorCluster, 0, 30, TimeUnit.SECONDS);
    }
    
    private void monitorCluster() {
        try {
            // 监控Broker状态
            monitorBrokers();
            
            // 监控Topic状态
            monitorTopics();
            
            // 监控消费者组
            monitorConsumerGroups();
            
        } catch (Exception e) {
            System.err.println("监控集群失败: " + e.getMessage());
        }
    }
    
    private void monitorBrokers() {
        try {
            DescribeClusterResult result = adminClient.describeCluster();
            Collection<Node> nodes = result.nodes().get(30, TimeUnit.SECONDS);
            
            System.out.println("=== Broker状态 ===");
            for (Node node : nodes) {
                System.out.println("Broker " + node.id() + ": " + node.host() + ":" + node.port());
            }
            
        } catch (Exception e) {
            System.err.println("监控Broker失败: " + e.getMessage());
        }
    }
    
    private void monitorTopics() {
        try {
            ListTopicsResult result = adminClient.listTopics();
            Set<String> topics = result.names().get(30, TimeUnit.SECONDS);
            
            System.out.println("=== Topic状态 ===");
            for (String topic : topics) {
                DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singletonList(topic));
                TopicDescription description = describeResult.topicNameValues().get(topic).get(30, TimeUnit.SECONDS);
                
                System.out.println("Topic: " + topic + ", 分区数: " + description.partitions().size());
            }
            
        } catch (Exception e) {
            System.err.println("监控Topic失败: " + e.getMessage());
        }
    }
    
    private void monitorConsumerGroups() {
        try {
            ListConsumerGroupsResult result = adminClient.listConsumerGroups();
            Collection<ConsumerGroupListing> groups = result.valid().get(30, TimeUnit.SECONDS);
            
            System.out.println("=== 消费者组状态 ===");
            for (ConsumerGroupListing group : groups) {
                System.out.println("消费者组: " + group.groupId() + ", 状态: " + group.state());
            }
            
        } catch (Exception e) {
            System.err.println("监控消费者组失败: " + e.getMessage());
        }
    }
}

// 性能指标监控
@Component
public class KafkaPerformanceMonitor {
    private final MeterRegistry meterRegistry;
    private final Timer producerTimer;
    private final Timer consumerTimer;
    private final Counter messageCounter;
    private final Gauge lagGauge;
    
    public KafkaPerformanceMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.producerTimer = Timer.builder("kafka.producer.latency")
            .register(meterRegistry);
        this.consumerTimer = Timer.builder("kafka.consumer.latency")
            .register(meterRegistry);
        this.messageCounter = Counter.builder("kafka.messages.processed")
            .register(meterRegistry);
        this.lagGauge = Gauge.builder("kafka.consumer.lag")
            .register(meterRegistry, this, KafkaPerformanceMonitor::getConsumerLag);
    }
    
    public void recordProducerLatency(Duration duration) {
        producerTimer.record(duration);
    }
    
    public void recordConsumerLatency(Duration duration) {
        consumerTimer.record(duration);
    }
    
    public void incrementMessageCount() {
        messageCounter.increment();
    }
    
    private double getConsumerLag() {
        // 这里可以集成Kafka的监控API获取消费延迟
        return 0.0;
    }
}

// 故障处理服务
@Service
public class KafkaFaultHandler {
    private final AdminClient adminClient;
    
    public KafkaFaultHandler(AdminClient adminClient) {
        this.adminClient = adminClient;
    }
    
    // 处理Broker故障
    public void handleBrokerFailure(int brokerId) {
        System.out.println("处理Broker " + brokerId + " 故障");
        
        try {
            // 检查Broker状态
            DescribeClusterResult result = adminClient.describeCluster();
            Collection<Node> nodes = result.nodes().get(30, TimeUnit.SECONDS);
            
            boolean brokerExists = nodes.stream()
                .anyMatch(node -> node.id() == brokerId);
            
            if (!brokerExists) {
                System.out.println("Broker " + brokerId + " 已下线");
                
                // 重新分配分区
                reassignPartitions(brokerId);
            }
            
        } catch (Exception e) {
            System.err.println("处理Broker故障失败: " + e.getMessage());
        }
    }
    
    // 重新分配分区
    private void reassignPartitions(int brokerId) {
        try {
            // 获取所有Topic
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topics = topicsResult.names().get(30, TimeUnit.SECONDS);
            
            // 重新分配分区
            for (String topic : topics) {
                reassignTopicPartitions(topic, brokerId);
            }
            
        } catch (Exception e) {
            System.err.println("重新分配分区失败: " + e.getMessage());
        }
    }
    
    // 重新分配Topic分区
    private void reassignTopicPartitions(String topic, int brokerId) {
        try {
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topic));
            TopicDescription description = result.topicNameValues().get(topic).get(30, TimeUnit.SECONDS);
            
            Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>();
            
            for (TopicPartitionInfo partitionInfo : description.partitions()) {
                if (partitionInfo.replicas().stream().anyMatch(node -> node.id() == brokerId)) {
                    // 重新分配分区
                    List<Integer> newReplicas = partitionInfo.replicas().stream()
                        .filter(node -> node.id() != brokerId)
                        .map(Node::id)
                        .collect(Collectors.toList());
                    
                    reassignments.put(new TopicPartition(topic, partitionInfo.partition()),
                        Optional.of(new NewPartitionReassignment(newReplicas)));
                }
            }
            
            if (!reassignments.isEmpty()) {
                AlterPartitionReassignmentsResult reassignResult = adminClient.alterPartitionReassignments(reassignments);
                reassignResult.all().get(30, TimeUnit.SECONDS);
                
                System.out.println("Topic " + topic + " 分区重新分配完成");
            }
            
        } catch (Exception e) {
            System.err.println("重新分配Topic分区失败: " + e.getMessage());
        }
    }
    
    // 处理分区不可用
    public void handlePartitionUnavailable(String topic, int partition) {
        System.out.println("处理分区不可用: " + topic + "-" + partition);
        
        try {
            // 检查分区状态
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topic));
            TopicDescription description = result.topicNameValues().get(topic).get(30, TimeUnit.SECONDS);
            
            TopicPartitionInfo partitionInfo = description.partitions().stream()
                .filter(p -> p.partition() == partition)
                .findFirst()
                .orElse(null);
            
            if (partitionInfo != null) {
                // 检查ISR状态
                if (partitionInfo.isr().isEmpty()) {
                    System.out.println("分区 " + topic + "-" + partition + " ISR为空");
                    
                    // 尝试恢复分区
                    recoverPartition(topic, partition);
                }
            }
            
        } catch (Exception e) {
            System.err.println("处理分区不可用失败: " + e.getMessage());
        }
    }
    
    // 恢复分区
    private void recoverPartition(String topic, int partition) {
        try {
            // 这里可以添加分区恢复逻辑
            System.out.println("尝试恢复分区: " + topic + "-" + partition);
            
        } catch (Exception e) {
            System.err.println("恢复分区失败: " + e.getMessage());
        }
    }
}

// 数据恢复策略
@Service
public class KafkaDataRecovery {
    private final AdminClient adminClient;
    
    public KafkaDataRecovery(AdminClient adminClient) {
        this.adminClient = adminClient;
    }
    
    // 恢复Topic数据
    public void recoverTopicData(String topic) {
        System.out.println("恢复Topic数据: " + topic);
        
        try {
            // 检查Topic状态
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topic));
            TopicDescription description = result.topicNameValues().get(topic).get(30, TimeUnit.SECONDS);
            
            // 检查每个分区的状态
            for (TopicPartitionInfo partitionInfo : description.partitions()) {
                if (partitionInfo.isr().size() < partitionInfo.replicas().size()) {
                    System.out.println("分区 " + partitionInfo.partition() + " 需要恢复");
                    
                    // 恢复分区
                    recoverPartition(topic, partitionInfo.partition());
                }
            }
            
        } catch (Exception e) {
            System.err.println("恢复Topic数据失败: " + e.getMessage());
        }
    }
    
    // 恢复分区
    private void recoverPartition(String topic, int partition) {
        try {
            // 这里可以添加分区恢复逻辑
            System.out.println("恢复分区: " + topic + "-" + partition);
            
        } catch (Exception e) {
            System.err.println("恢复分区失败: " + e.getMessage());
        }
    }
    
    // 备份Topic数据
    public void backupTopicData(String topic, String backupPath) {
        System.out.println("备份Topic数据: " + topic + " 到 " + backupPath);
        
        try {
            // 这里可以添加数据备份逻辑
            System.out.println("Topic数据备份完成");
            
        } catch (Exception e) {
            System.err.println("备份Topic数据失败: " + e.getMessage());
        }
    }
    
    // 从备份恢复数据
    public void restoreTopicData(String topic, String backupPath) {
        System.out.println("从备份恢复Topic数据: " + topic + " 从 " + backupPath);
        
        try {
            // 这里可以添加数据恢复逻辑
            System.out.println("Topic数据恢复完成");
            
        } catch (Exception e) {
            System.err.println("恢复Topic数据失败: " + e.getMessage());
        }
    }
}