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

    public static void main(String[] args) {
        // 消费者组ID，可以相同也可以不同，取决于业务逻辑
        String groupId = "my-processing-group";
        String bootstrapServers = "localhost:9092"; // Kafka集群地址

        // --- 消费者实例1：处理常规主题，使用默认或较短的max.poll.interval.ms ---
        Properties defaultProps = new Properties();
        defaultProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        defaultProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        defaultProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        defaultProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 设置常规的 max.poll.interval.ms，例如 5 分钟 (300000 毫秒)
        defaultProps.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000"); 

        KafkaConsumer<String, String> defaultConsumer = new KafkaConsumer<>(defaultProps);
        // 订阅常规处理的主题
        defaultConsumer.subscribe(Collections.singletonList("topic-fast-processing"));
        System.out.println("消费者实例1 (常规处理) 已订阅 'topic-fast-processing'，max.poll.interval.ms: 300000ms");

        // --- 消费者实例2：处理需要长时间处理的主题，使用更长的max.poll.interval.ms ---
        Properties longProcessingProps = new Properties();
        longProcessingProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        longProcessingProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId); // 可以与实例1在同一个消费者组
        longProcessingProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        longProcessingProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 针对特定主题设置更长的 max.poll.interval.ms，例如 30 分钟 (1800000 毫秒)
        longProcessingProps.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "1800000"); 

        KafkaConsumer<String, String> longProcessingConsumer = new KafkaConsumer<>(longProcessingProps);
        // 订阅需要长时间处理的主题
        longProcessingConsumer.subscribe(Collections.singletonList("topic-long-processing"));
        System.out.println("消费者实例2 (长时间处理) 已订阅 'topic-long-processing'，max.poll.interval.ms: 1800000ms");

        // 在实际应用中，这两个消费者实例通常会在独立的线程或进程中启动，
        // 并持续调用 poll() 方法来拉取和处理消息。
        // 例如：
        // new Thread(() -> {
        //     try {
        //         while (true) {
        //             ConsumerRecords<String, String> records = defaultConsumer.poll(Duration.ofMillis(100));
        //             // 处理 records...
        //             defaultConsumer.commitSync();
        //         }
        //     } catch (Exception e) { e.printStackTrace(); } finally { defaultConsumer.close(); }
        // }).start();
        //
        // new Thread(() -> {
        //     try {
        //         while (true) {
        //             ConsumerRecords<String, String> records = longProcessingConsumer.poll(Duration.ofMillis(100));
        //             // 处理 records... 可能耗时较长
        //             longProcessingConsumer.commitSync();
        //         }
        //     } catch (Exception e) { e.printStackTrace(); } finally { longProcessingConsumer.close(); }
        // }).start();

        System.out.println("\n消费者初始化完成。在生产环境中，请确保每个消费者实例在独立的执行流中进行消息轮询和处理。");
    }
}