Kafka max.poll.interval.ms配置详解及按主题隔离策略

import org.apache.kafka.clients.consumer.ConsumerRecords;
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) {
        // 配置用于处理 'topic-long-processing' 的消费者实例
        Properties longProcessingProps = new Properties();
        longProcessingProps.put("bootstrap.servers", "localhost:9092");
        longProcessingProps.put("group.id", "long-processing-group");
        longProcessingProps.put("key.deserializer", StringDeserializer.class.getName());
        longProcessingProps.put("value.deserializer", StringDeserializer.class.getName());
        // 为耗时主题设置较长的 max.poll.interval.ms (例如 15 分钟)
        longProcessingProps.put("max.poll.interval.ms", "900000"); // 15 * 60 * 1000 ms

        KafkaConsumer<String, String> longProcessingConsumer = new KafkaConsumer<>(longProcessingProps);
        longProcessingConsumer.subscribe(Collections.singletonList("topic-long-processing"));

        // 配置用于处理 'topic-short-processing' 的消费者实例
        Properties shortProcessingProps = new Properties();
        shortProcessingProps.put("bootstrap.servers", "localhost:9092");
        shortProcessingProps.put("group.id", "short-processing-group"); // 可以是不同的消费者组
        shortProcessingProps.put("key.deserializer", StringDeserializer.class.getName());
        shortProcessingProps.put("value.deserializer", StringDeserializer.class.getName());
        // 为快速处理主题设置默认或较短的 max.poll.interval.ms (例如 1 分钟)
        shortProcessingProps.put("max.poll.interval.ms", "60000"); // 1 * 60 * 1000 ms

        KafkaConsumer<String, String> shortProcessingConsumer = new KafkaConsumer<>(shortProcessingProps);
        shortProcessingConsumer.subscribe(Collections.singletonList("topic-short-processing"));

        // 在不同的线程或进程中运行这两个消费者
        // 消费者1线程：处理 long-processing-consumer
        new Thread(() -> {
            try {
                while (true) {
                    ConsumerRecords<String, String> records = longProcessingConsumer.poll(Duration.ofMillis(100));
                    records.forEach(record -> {
                        System.out.println("Long Processing: " + record.value());
                        // 模拟长时间处理
                        try {
                            Thread.sleep(5 * 60 * 1000); // 模拟处理 5 分钟
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    });
                    longProcessingConsumer.commitSync(); // 提交偏移量
                }
            } finally {
                longProcessingConsumer.close();
            }
        }).start();

        // 消费者2线程：处理 short-processing-consumer
        new Thread(() -> {
            try {
                while (true) {
                    ConsumerRecords<String, String> records = shortProcessingConsumer.poll(Duration.ofMillis(100));
                    records.forEach(record -> {
                        System.out.println("Short Processing: " + record.value());
                        // 模拟短时间处理
                        try {
                            Thread.sleep(500); // 模拟处理 0.5 秒
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    });
                    shortProcessingConsumer.commitSync(); // 提交偏移量
                }
            } finally {
                shortProcessingConsumer.close();
            }
        }).start();
    }
}