【spring boot】 重启kafka客户端连接

背景

kafka 服务端重建时，kafka客户端会连不上kafka服务端，此时需要重启客户端重连

代码实现

@Service
public class KafkaConsumerService {
    private KafkaConsumer<String, String> consumer;
    
    @Autowired
    private KafkaProperties kafkaProperties;

   //在应用程序启动时初始化Kafka消费者，并启动一个单独的线程用于消费消息
    @PostConstruct
    public void init() {
        consumer = new KafkaConsumer<>(kafkaProperties.buildConsumerProperties());
        consumer.subscribe(Arrays.asList("myTopic"));
        Executors.newSingleThreadExecutor().submit(this::consume);
    }

    @PreDestroy
    public void destroy() {
        consumer.close();
    }

    private void consume() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 处理消息逻辑
            }
        }
    }

   //先关闭现有的Kafka消费者，然后创建一个新的消费者实例，并重新订阅我们的主题。最后，我们使用 ExecutorService 启动一个新的消费者线程
    public void restart() {
        consumer.close();
        consumer = new KafkaConsumer<>(kafkaProperties.buildConsumerProperties());
        consumer.subscribe(Arrays.asList("myTopic"));
        Executors.newSingleThreadExecutor().submit(this::consume);
    }
}