// ConsumerConfig.java
package com.example.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

public class ConsumerConfig {
    
    /**
     * 配置消费者
     */
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // 性能优化配置
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 每次fetch最小1KB
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // 最大等待500ms
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 10485760); // 10MB
        
        // 并发配置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); // 每次poll最多100条
        
        // 自动提交配置
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        
        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    /**
     * 配置监听器容器工厂
     * 增加并发消费者数量
     */
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        
        // 设置并发消费者数量
        factory.setConcurrency(10); // 10个并发消费者
        
        // 设置批处理
        factory.setBatchListener(true);
        
        // 设置手动提交
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        return factory;
    }
}

// OptimizedConsumer.java
package com.example.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.annotation.Async;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class OptimizedConsumer {
    
    private ExecutorService executorService = Executors.newFixedThreadPool(20);
    
    /**
     * 优化1: 批量处理
     */
    @KafkaListener(topics = "my-topic", concurrency = "10")
    public void batchConsume(List<ConsumerRecord<String, String>> records, 
                             Acknowledgment ack) {
        // 批量处理消息
        for (ConsumerRecord<String, String> record : records) {
            processMessage(record);
        }
        
        // 手动提交
        ack.acknowledge();
    }
    
    /**
     * 优化2: 异步处理
     */
    @KafkaListener(topics = "my-topic")
    public void asyncConsume(ConsumerRecord<String, String> record) {
        // 提交异步任务
        CompletableFuture.runAsync(() -> {
            processMessage(record);
        }, executorService);
    }
    
    /**
     * 优化3: 本地缓存批量处理
     */
    @KafkaListener(topics = "my-topic")
    public void cachedConsume(ConsumerRecord<String, String> record) {
        // 将消息缓存到本地
        localCache.add(record);
        
        // 批量达到阈值时处理
        if (localCache.size() >= 100) {
            List<ConsumerRecord<String, String>> batch = 
                new ArrayList<>(localCache);
            localCache.clear();
            
            // 异步处理批量
            executorService.submit(() -> processBatch(batch));
        }
    }
    
    /**
     * 优化4: 消费逻辑优化
     */
    private void processMessage(ConsumerRecord<String, String> record) {
        try {
            // 1. 快速预处理
            String key = record.key();
            String value = record.value();
            
            // 2. 避免重复处理（幂等性）
            if (isProcessed(key)) {
                return;
            }
            
            // 3. 使用连接池
            processWithConnectionPool(value);
            
            // 4. 标记已处理
            markAsProcessed(key);
            
        } catch (Exception e) {
            log.error("处理消息失败", e);
            // 发送到死信队列
            sendToDlq(record);
        }
    }
    
    /**
     * 优化5: 预取优化
     */
    public void optimizedPoll(KafkaConsumer<String, String> consumer) {
        while (true) {
            // 增加fetch.max.wait.ms
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofSeconds(5));
            
            if (records.isEmpty()) {
                continue;
            }
            
            // 批量处理
            processBatch(records);
            
            // 手动提交
            consumer.commitSync();
        }
    }
}

// PerformanceTest.java
package com.example.kafka.test;

import org.junit.Test;

public class PerformanceTest {
    
    /**
     * 消费能力测试
     */
    @Test
    public void testConsumerThroughput() {
        // 1. 生产测试数据
        produceTestData(10000);
        
        // 2. 测试消费速率
        long startTime = System.currentTimeMillis();
        consumeMessages();
        long endTime = System.currentTimeMillis();
        
        long duration = endTime - startTime;
        double throughput = 10000.0 / (duration / 1000.0);
        
        System.out.println("消费速率: " + throughput + " msg/s");
        
        // 3. 判断是否满足需求
        assert throughput > 1000 : "消费速率不足";
    }
    
    /**
     * 积压恢复测试
     */
    @Test
    public void testLagRecovery() {
        // 1. 模拟积压
        produceTestData(100000);
        
        // 2. 测试恢复时间
        long startTime = System.currentTimeMillis();
        
        // 启动消费者
        startConsumers(10);
        
        // 等待消费完成
        waitForConsumptionComplete();
        
        long endTime = System.currentTimeMillis();
        
        System.out.println("恢复时间: " + (endTime - startTime) / 1000 + "s");
    }
}