背景
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);
}
}
标签:consume,spring,void,boot,kafka,consumer,public,客户端
From: https://www.cnblogs.com/xiaoyu-jane/p/17345762.html