kafka的连接问题,我如何通过代码知道服务端的kafka服务是否开启？

Question

我在电脑的虚拟机搭建了kafka服务,在本地使用Java客户端进行访问,现在假如虚拟机上的服务被我关闭,我在本地的代码就会无法去向kafka推送消息，并且会在1分钟后报一个timeout的错,我可以控制这个timeout的时间吗？我应该怎么去知道服务端的kafka服务处在active状态？下面是我的本地的客户端代码,像部署在虚拟机的kafka服务端推送消息，假如我将服务端的kafka服务关闭，或者直接打开服务端的防火墙，程序就会卡着，直到报错：

org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.88.131: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);
        for (int i = 0; i < 100; i++){
//          Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("topicTe1st", Integer.toString(i), Integer.toString(i)));
            producer.send(new ProducerRecord<String, String>("topicTest", Integer.toString(i), Integer.toString(i)),
                                   new Callback() {
                        public void onCompletion(RecordMetadata metadata, Exception e) {
                            if(e != null) {
                               e.printStackTrace();
                            } else {
                               System.out.println("The offset of the record we just sent is: " + metadata.partition());
                          }
                        }
                    });
            //System.out.println(future.get().toString());
            System.out.println("send one ");
        }
        producer.close();
    }

那我应该如果通过代码去事先就知道kafka服务是否开启着？

Answer 1

不要把一个程序做的大而全，分而治之的思想。服务监控不需要放在你的程序代码中去考虑。你的程序只需要做到两点:

• 连接kafka有超时设置
• 失败能重试

以上两点用kafka官方client都有配置，可以直接配置使用。

至于kafka服务交给服务监控系统去处理。最简单的一种策略是把kafka做成系统服务(CDH版的天生自带)，交由系统的服务管理系统去管理，用service kafka status就能看到服务运行状态，如果有报警系统，交给报警系统去报警，完了。

你的程序不应该越俎代庖，去把服务管理的活给干了。