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");
        // 消費者組唯一標(biāo)識
        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());
 
        // 消費者調(diào)優(yōu)參數(shù)設(shè)置
        // 批量拉取消息的最小字節(jié)數(shù)，優(yōu)化拉取效率
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "50000");
        // 批量消息拉取的最大等待時間
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100");
        // 每次poll的最大消息數(shù)，防止單次處理時間過長
        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 {
                // 在關(guān)閉前同步提交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");
        // 針對高負載做了批量處理和多線程優(yōu)化
        props.put("max.poll.records", "1000");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("high-load-topic"));
 
        // 創(chuàng)建線程池并發(fā)處理消息
        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) {
                        // 進行業(yè)務(wù)邏輯處理
                        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();
        }
    }
}