深入探讨Kafka消费者高性能调优与实践

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
 
import java.util.Collections;
import java.util.Properties;
 
public class OptimizedKafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 指定Kafka集群的地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 消费者组唯一标识
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "optimized-consumer-group");
        // 禁用自动提交offset，采用手动或异步提交策略
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 反序列化器
        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, "50000");
        // 批量消息拉取的最大等待时间
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100");
        // 每次poll的最大消息数，防止单次处理时间过长
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
        // 会话超时时间配置，保证消费者心跳机制的正常运行
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("optimized-topic"));
 
        try {
            while (true) {
                // 轮询获取消息
                var records = consumer.poll(java.time.Duration.ofMillis(100));
                records.forEach(record -> {
                    // 处理消费到的消息
                    System.out.println("offset = " + record.offset() + ", key = " + record.key() + ", value = " + record.value());
                });
                // 异步提交offset，提升性能
                consumer.commitAsync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                // 在关闭前同步提交offset，防止消息丢失
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
 
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class ConcurrentKafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "concurrent-consumer-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 针对高负载做了批量处理和多线程优化
        props.put("max.poll.records", "1000");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("high-load-topic"));
 
        // 创建线程池并发处理消息
        ExecutorService executor = Executors.newFixedThreadPool(4);
 
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                // 将每批消息分配到线程池中处理
                executor.submit(() -> {
                    for (ConsumerRecord<String, String> record : records) {
                        // 进行业务逻辑处理
                        System.out.printf("Thread: %s, Offset: %d, Key: %s, Value: %s%n",
                                Thread.currentThread().getName(), record.offset(), record.key(), record.value());
                    }
                });
                // 异步提交offset，减少同步阻塞
                consumer.commitAsync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.commitSync();
            consumer.close();
            executor.shutdown();
        }
    }
}