我明白你的需求——默认的@KafkaListener是持续保持连接、实时消费消息的，而你希望改成定期轮询的模式，比如每隔一段时间才去Kafka拉取一次消息，平时不监听。下面结合你的现有代码，给你几种可行的实现方案：

方案一：通过定时任务控制消费者容器的启停/暂停恢复 #

这种方式利用Spring Kafka提供的KafkaListenerEndpointRegistry管理消费者容器，配合Spring定时任务，定期启动消费，完成后暂停，完美契合你想要的"轮询"效果。

步骤1：修改你的消费者服务类 #

给@KafkaListener指定唯一ID，注入容器注册表，添加定时任务控制容器状态：

@Service
public class KafkaReciever {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaReciever.class);
    private CountDownLatch latch = new CountDownLatch(1);
    
    // 注入容器注册表，用来精准管理消费者容器
    @Autowired
    private KafkaListenerEndpointRegistry registry;

    public CountDownLatch getLatch() {
        return latch;
    }

    // 给消费者指定id，方便后续定位容器
    @KafkaListener(topics = "test", id = "testConsumer") 
    public void receive(String payload) {
        LOGGER.info("received payload='{}'", payload);
        latch.countDown();
    }

    // 定时任务：示例为每隔5分钟执行一次消费（300000毫秒）
    @Scheduled(fixedRate = 300000) 
    public void pollKafkaPeriodically() {
        // 获取指定id的消费者容器
        MessageListenerContainer container = registry.getListenerContainer("testConsumer");
        
        // 启动/恢复容器：如果容器未运行则启动，已运行但暂停则恢复
        if (!container.isRunning()) {
            container.start();
        } else if (container.isPaused()) {
            container.resume();
        }
        
        // 等待消息处理完成（根据你的业务耗时调整等待时间）
        try {
            // 最多等待10秒，确保消息处理完成
            if (latch.await(10, TimeUnit.SECONDS)) {
                LOGGER.info("本次轮询消费完成");
            }
            latch = new CountDownLatch(1); // 重置计数器，为下次轮询做准备
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LOGGER.error("等待消费完成时被中断", e);
        }
        
        // 消费完成后暂停容器，停止实时监听
        container.pause();
    }
}

步骤2：开启定时任务支持 #

在你的Kafka配置类上添加@EnableScheduling注解，启用Spring定时任务：

@Configuration
@EnableScheduling // 必须添加这个注解才能生效定时任务
public class KafkaConsumerConfig {
    // 你的现有配置代码...
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "foo1");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 可选：限制每次轮询拉取的最大消息数
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        return props;
    }

    // 其他Bean定义（consumerFactory、kafkaListenerContainerFactory）保持不变...
}

注意事项 #

• 务必给@KafkaListener指定id，否则无法通过注册表找到对应的容器。
• latch.await()的等待时间需要根据你的消息处理耗时调整，确保所有消息处理完成后再暂停容器。
• 暂停容器后，消费者会停止主动拉取消息，直到下次定时任务恢复它。

方案二：手动创建消费者，定时调用poll方法 #

如果你想要完全掌控消费流程，可以脱离@KafkaListener，自己手动创建消费者，定时调用poll()方法拉取消息：

@Service
public class KafkaReciever {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaReciever.class);
    private final Consumer<String, String> consumer;

    // 通过构造函数注入ConsumerFactory，手动创建消费者
    public KafkaReciever(ConsumerFactory<String, String> consumerFactory) {
        this.consumer = consumerFactory.createConsumer();
        // 订阅目标主题
        consumer.subscribe(Collections.singletonList("test"));
    }

    // 定时轮询：示例为每隔5分钟拉取一次消息
    @Scheduled(fixedRate = 300000) 
    public void pollKafka() {
        try {
            // 拉取消息，超时时间设为1秒（没有消息就立即返回）
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                LOGGER.info("received payload='{}' | partition: {} | offset: {}", 
                            record.value(), record.partition(), record.offset());
                // 手动提交偏移量（如果需要精确控制偏移量的话）
                consumer.commitSync();
            }
        } catch (WakeupException e) {
            // 唤醒异常是关闭消费者时的正常操作，无需处理
        } catch (Exception e) {
            LOGGER.error("轮询消息失败", e);
        }
    }

    // 应用关闭时，优雅关闭消费者
    @PreDestroy
    public void closeConsumer() {
        consumer.wakeup();
        consumer.close();
    }
}

这种方式的优点是完全掌控消费时机和流程，缺点是需要自己处理偏移量提交、消费者关闭等细节，适合对消费逻辑有高度自定义需求的场景。

方案三：调整消费者配置（伪轮询，不推荐） #

如果你只是想控制消费的间隔，而非完全停止监听，可以调整以下配置，但这并不是真正的"定期轮询"，消费者仍会保持与Kafka的连接：

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    // 其他现有配置...
    // 限制每次拉取的最大消息数
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    // 设置两次拉取之间的最大间隔时间（示例为5分钟）
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
    return props;
}

这种方式下，消费者每次拉取消息后，会等待到max.poll.interval.ms设定的时间再拉取下一次，但始终保持连接，不适合需要完全停止监听的场景。

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