kafka如何判断连接是否可以使用java实例代码

在Java中，可以通过使用Kafka的Java客户端来判断Kafka连接是否可用。以下是一个简单的Java示例代码，可以用来检查Kafka连接是否可用：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.errors.TimeoutException;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class KafkaConnectionChecker {

    private static final String TOPIC_NAME = "test-topic";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 检查Broker可用性
            if (adminClient.describeCluster().nodes().get().isEmpty()) {
                System.err.println("No brokers found in Kafka cluster");
                return;
            }

            // 检查Topic可用性
            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(TOPIC_NAME));
            try {
                KafkaFuture<TopicDescription> topicDescriptionFuture = describeTopicsResult.values().get(TOPIC_NAME);
                topicDescriptionFuture.get(5000, TimeUnit.MILLISECONDS);
            } catch (TimeoutException e) {
                System.err.println("Timed out while waiting for topic metadata");
                return;
            } catch (InterruptedException | ExecutionException e) {
                System.err.println("Failed to get topic metadata: " + e.getMessage());
                return;
            }

            // 输出连接成功信息
            System.out.println("Kafka connection is available");
        } catch (Exception e) {
            System.err.println("Failed to check Kafka connection: " + e.getMessage());
        }
    }
}

上述代码通过创建一个`AdminClient`实例，并使用`describeCluster()`和`describeTopics()`方法来检查Kafka连接的可用性。如果连接成功，则输出一条连接成功的信息。如果连接失败，则输出相应的错误信息。

需要注意的是，上述代码仅仅是一个简单的示例，实际应用中还需要考虑更多的因素来保证Kafka连接的稳定性和可靠性。例如，需要实现重试机制、异常处理、监控和报警等。