在kafka單伺服器的情況下消費者端如何確定伺服器是否啟動

在kafka集群的單伺服器情況下，如何通過consumer消費者確定kafka伺服器或者zookeeper伺服器是否啟動(因為消費者目前是無法判斷伺服器是否啟動的，它只是去輪詢獲取伺服器數據而不報錯)，如果沒有啟動，消費者端做出相應的操作來提醒消費者端使用人員進行維護，在這裡我提供一個簡單的解決方案，可能並不是非常通用，提供一個簡單的思路而已。

如果kafka伺服器或者zookeeper伺服器沒有啟動，在producer生產者端向伺服器發送信息的時候會出現錯誤(org.apache.kafka.common.errors.TimeoutException)，我使用的kafka的maven版本是0.11.0.0的，當有異常錯誤出現的時候我們可以在生產者端使用一個靜態變數來變更記錄這個狀態，同時在生產者端提供一個介面以供消費者端調用；在consumer消費者端我們提供一個定時任務，如果消費者端在規定的時間裡面沒有從kafka伺服器獲取到數據，定義一個靜態變數去記錄獲取數據狀態，那麼在定時任務裡面就會根據這個靜態變數的值決定是否去調用producer消費者端的介面判斷介面返回的狀態情況，如果返回的狀態表示伺服器沒有啟動則消費者端做出相應的操作。如果在集群多伺服器的情況下，客戶端可以根據訂閱的主題topic判斷哪些伺服器是否存活的，下面提供了生產者producer和消費者consumer的main方法，並沒有按照上面的敘述去完成介面，如果需要自己去實現。

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.11.0.0</version>
</dependency>

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.KafkaProducer;

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

/**
* @author liaoyubo
* @version 1.0 2017/7/25
* @description
*/
public class KafkaProducerTest {

public static void main(String [] args){
Properties properties = new Properties;
//properties.put("zookeeper.connect","localhost:2181");
properties.put("bootstrap.servers", "192.168.201.190:9092");
properties.put("acks", "all");
properties.put("retries", 0);
properties.put("batch.size", 16384);
properties.put("linger.ms", 1);
properties.put("buffer.memory", 33554432);
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String,String> producer = new KafkaProducer<String, String>(properties);

for(int i = 0;i < 10;i++){
Future<RecordMetadata> futureRecordMetadata = producer.send(new ProducerRecord<String, String>("myTopic",Integer.toString(i),Integer.toString(i)));
try {
futureRecordMetadata.get(3, TimeUnit.SECONDS);
System.out.println("發送的message:"+i);
} catch (InterruptedException e) {
e.printStackTrace;
} catch (ExecutionException e) {
if(e.getMessage.split(":")[0].split("\.")[5].equals("TimeoutException")){
System.out.println("無法連接到伺服器");
}
e.printStackTrace;
} catch (TimeoutException e) {
System.out.println("無法連接到伺服器");
e.printStackTrace;
}
}

producer.close;

}

}

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
* @author liaoyubo
* @version 1.0 2017/7/26
* @description
*/
public class KafkaConsumerClientTest {

public static void main(String [] args){

Properties properties = new Properties;
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.201.190:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

Consumer<String,String> consumer = new KafkaConsumer<String, String>(properties);

consumer.subscribe(Arrays.asList("myTopic","myTest"));
while (true){
ConsumerRecords<String,String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records){
//int partition = record.partition;
String topic = record.topic;
List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
for (PartitionInfo partitionInfo : partitionInfoList){
Node node = partitionInfo.leader;
System.out.println(node.host);
//獲取存活的伺服器
Node nodes = partitionInfo.replicas;
}
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset, record.key, record.value);
}
}
}

}

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！