实现自定义Kafka Consumer：移除while循环的方案求助

嘿，我来给你几个靠谱的方案解决这个问题！

如果你的项目是基于Spring生态的，那Spring Kafka绝对是最优解——它帮你封装好了所有线程管理、持续监听的逻辑，完全不用自己写while循环，只要加个注解就能实现自动触发消息处理。

• 先确保引入Spring Kafka依赖（以Maven为例）：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>选和你的Spring版本兼容的版本</version>
</dependency>

• 配置消费者参数（比如在application.yml里）：

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092 # 你的Kafka地址
      group-id: your-consumer-group-id # 自定义消费者组ID
      auto-offset-reset: latest # 只消费最新消息
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

• 最后写你的消息处理方法，用@KafkaListener标记要监听的主题：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class YourMessageHandler {

    // 只要主题有新消息，这个方法就会自动被调用
    @KafkaListener(topics = "your-target-topic")
    public void customConsumer(String message) {
        // 这里写你原来的消息处理逻辑
        System.out.println("收到新消息啦：" + message);
    }
}

Spring容器启动后，会自动创建并维护消费者线程，持续监听指定主题，完全不用你操心循环和线程的事儿。

如果你的项目没用到Spring，那可以把消费者的监听逻辑封装到独立线程里，替代原来的while循环，实现自动持续监听：

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerListener {
    private final KafkaConsumer<String, String> consumer;
    private final String targetTopic;
    private volatile boolean keepRunning = true;

    // 构造方法初始化消费者并启动监听线程
    public KafkaConsumerListener(Properties consumerProps, String topic) {
        this.consumer = new KafkaConsumer<>(consumerProps);
        this.targetTopic = topic;
        consumer.subscribe(Collections.singletonList(topic));
        // 启动专属监听线程
        new Thread(this::startListening, "kafka-listener-thread").start();
    }

    // 监听逻辑封装在这里
    private void startListening() {
        while (keepRunning) {
            // 拉取消息（超时时间可自定义）
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            // 遍历消息并调用你的处理方法
            records.forEach(record -> customConsumer(record.value()));
        }
        // 停止时关闭消费者
        consumer.close();
    }

    // 你的消息处理逻辑，有新消息就会触发
    private void customConsumer(String message) {
        System.out.println("正在处理消息：" + message);
    }

    // 提供优雅停止的方法
    public void stopListening() {
        keepRunning = false;
    }
}

使用的时候只需要初始化这个类就行：

public class Main {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "your-group-id");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 初始化后自动开始监听
        KafkaConsumerListener listener = new KafkaConsumerListener(props, "your-target-topic");
        
        // 要停止监听时调用：listener.stopListening();
    }
}

你之前尝试ExecutorService没成功，大概率是没让任务持续执行——比如你可能只提交了一次单次任务，而不是一个持续运行的任务。其实可以把上面的startListening方法提交到线程池里，比如：

ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(listener::startListening);

这样线程池会维持这个线程一直运行，只要keepRunning为true，就会持续监听主题。

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