Kafka消费者批量拉取优化：基于字节大小动态控制消息数量

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaByteLimitedConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-byte-limited-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 核心配置：设置每次拉取最大字节数，例如1MB
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1 * 1024 * 1024); // 1MB

        // 辅助配置：将MAX_POLL_RECORDS_CONFIG设置为一个非常大的值，确保字节限制优先
        // 假设平均消息50B，1MB可容纳20480条，设置20万确保不会先达到记录数限制
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200000); 

        // 其他重要配置
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 禁用自动提交，手动控制提交
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的偏移量开始消费

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_topic"));

        try {
            while (true) {
                // poll方法会根据配置拉取消息
                // 实际返回的记录数和总字节数将受到这两个参数的共同影响，但FETCH_MAX_BYTES_CONFIG将是主要限制
                var records = consumer.poll(Duration.ofMillis(100));
                if (!records.isEmpty()) {
                    System.out.println("Pulled " + records.count() + " records. Total bytes (approx): " + calculateApproximateBytes(records) + " bytes.");
                    // 处理消息
                    records.forEach(record -> {
                        // System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                    });
                    consumer.commitAsync(); // 手动异步提交偏移量
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }