Kafka中关闭broker时消费者无法工作的问题。

首先，需要确保消费者在订阅主题时，设置合适的组ID。 假设我们的消费者实例是“Consumer”，组ID设置为“group1”：

consumerProperties.put("group.id", "group1"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);

然后，在消费者中启用自动提交偏移量设置，以确保在关闭Kafka broker时，消费者提交偏移量并在重启后恢复其状态。 我们可以通过以下方式确保设置：

consumerProperties.put("enable.auto.commit", "true"); consumerProperties.put("auto.commit.interval.ms", "1000");

此外，我们还可以创建一个shutdown hook，以确保在关闭应用程序时正确关闭Kafka消费者。

Runtime.getRuntime().addShutdownHook(new Thread(() -> { logger.info("Closing Kafka consumer on shutdown"); consumer.close(); }));

使用这些方法，我们可以确保在关闭Kafka broker时，消费者将提交偏移量，并在重启后恢复其状态。