Java Kafka Consumer轮询超时设置：60分钟后停止轮询的实现方案

Kafka Consumer 没有内置的配置属性可以直接设置轮询超时停止，必须通过编程方式来控制运行时长。以下是几种常用的实现方案：

1. 基于启动时间的时长判断 #

记录 Consumer 的启动时间，每次轮询后检查已运行时长是否达到60分钟，达到则停止轮询并关闭 Consumer。这种方式简单直接，适合大多数场景。

示例代码：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;

public class TimedKafkaConsumer {
    public static void main(String[] args) {
        // 初始化 Kafka Consumer（此处省略配置构建逻辑）
        Consumer<String, String> consumer = ...;
        consumer.subscribe(Arrays.asList("target-topic"));

        final long RUN_DURATION_MINUTES = 60;
        Instant startTime = Instant.now();
        boolean keepPolling = true;

        while (keepPolling) {
            // 轮询消息，设置较短的超时时间以便及时检查停止条件
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            // 处理获取到的消息
            processMessages(records);

            // 计算已运行时长，判断是否停止
            long elapsedMinutes = Duration.between(startTime, Instant.now()).toMinutes();
            if (elapsedMinutes >= RUN_DURATION_MINUTES) {
                keepPolling = false;
            }
        }

        // 关闭 Consumer，释放资源
        consumer.close(Duration.ofSeconds(10));
    }

    private static void processMessages(ConsumerRecords<String, String> records) {
        // 自定义消息处理逻辑
        records.forEach(record -> {
            System.out.printf("Received message: key=%s, value=%s%n", record.key(), record.value());
        });
    }
}

2. 使用定时任务触发停止 #

通过 ScheduledExecutorService 调度一个定时任务，在60分钟后设置停止标志，轮询循环中检查该标志来决定是否继续运行。这种方式适合需要更灵活控制停止时机的场景。

示例代码：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledStopConsumer {
    public static void main(String[] args) {
        Consumer<String, String> consumer = ...;
        consumer.subscribe(Arrays.asList("target-topic"));

        volatile boolean stopPolling = false;
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

        // 60分钟后触发停止标志
        scheduler.schedule(() -> stopPolling = true, 60, TimeUnit.MINUTES);

        while (!stopPolling) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            processMessages(records);
        }

        // 关闭资源
        consumer.close(Duration.ofSeconds(10));
        scheduler.shutdown();
    }

    private static void processMessages(ConsumerRecords<String, String> records) {
        // 消息处理逻辑
    }
}

注意事项 #

• 轮询的超时时间（poll 方法的参数）不要设置过长，否则即使达到停止时间，Consumer 仍会阻塞在 poll 调用中，无法及时停止。建议设置为1秒以内的较短时长。
• 停止时务必调用 consumer.close() 方法，确保 Consumer 正常提交偏移量并释放连接资源。

内容的提问来源于stack exchange，提问作者Ananya