Kafka集群不可用，如何监听重新连接？

1. 在Kafka的配置文件中，设置如下参数以启用自动重试：

retry.backoff.ms=500
reconnect.backoff.ms=1000

2. 在生产者和消费者代码中，使用try-catch块处理Kafka连接异常，并进行重试，例如：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
try {
    producer.send(new ProducerRecord<>("my_topic", "my_key", "my_value"));
} catch (Exception e) {
    // 处理连接异常，等待一段时间后重试
    Thread.sleep(1000);
    producer.send(new ProducerRecord<>("my_topic", "my_key", "my_value"));
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my_group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
try {
    consumer.subscribe(Arrays.asList("my_topic"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.key() + ":" + record.value());
        }
    }
} catch (Exception e) {
    // 处理连接异常，等待一段时间后重新订阅
    Thread.sleep(1000);
    consumer.subscribe(Arrays.asList("my_topic"));