Kafka消费者max.poll.interval.ms参数详解与主题隔离实践

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 TopicSpecificMaxPollInterval {

    public static void main(String[] args) {
        // 1. 通用消费者配置 - 用于处理常规消息
        Properties commonConsumerProps = new Properties();
        commonConsumerProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        commonConsumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "common_consumer_group");
        commonConsumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        commonConsumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 设置一个相对较短的max.poll.interval.ms，例如30秒
        commonConsumerProps.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "30000"); // 30 seconds

        KafkaConsumer<String, String> commonConsumer = new KafkaConsumer<>(commonConsumerProps);
        String commonTopic = "general_topic";
        commonConsumer.subscribe(Collections.singletonList(commonTopic));
        System.out.println("Common Consumer subscribed to: " + commonTopic + " with max.poll.interval.ms=" + commonConsumerProps.getProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG));

        // 2. 特定主题消费者配置 - 用于处理耗时较长的消息
        Properties specialTopicConsumerProps = new Properties();
        specialTopicConsumerProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        specialTopicConsumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "special_consumer_group"); // 不同的消费者组ID或相同的组ID但不同的实例
        specialTopicConsumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        specialTopicConsumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 设置一个较长的max.poll.interval.ms，例如5分钟
        specialTopicConsumerProps.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000"); // 5 minutes

        KafkaConsumer<String, String> specialTopicConsumer = new KafkaConsumer<>(specialTopicConsumerProps);
        String specialTopic = "long_processing_topic";
        specialTopicConsumer.subscribe(Collections.singletonList(specialTopic));
        System.out.println("Special Consumer subscribed to: " + specialTopic + " with max.poll.interval.ms=" + specialTopicConsumerProps.getProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG));

        // 模拟消费者持续运行和消息处理
        // 在实际应用中，这里会有一个循环来调用 consumer.poll()
        // 并在处理完消息后提交位移
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Closing consumers...");
            commonConsumer.close();
            specialTopicConsumer.close();
        }));

        // 实际的poll循环和消息处理逻辑
        // while (true) {
        //     ConsumerRecords<String, String> records = commonConsumer.poll(Duration.ofMillis(100));
        //     for (ConsumerRecord<String, String> record : records) {
        //         // 处理 commonTopic 的消息
        //     }
        //     commonConsumer.commitSync();
        //
        //     records = specialTopicConsumer.poll(Duration.ofMillis(100));
        //     for (ConsumerRecord<String, String> record : records) {
        //         // 处理 specialTopic 的消息，可能耗时更长
        //     }
        //     specialTopicConsumer.commitSync();
        // }
    }
}