動態從zookeeper中讀取kafka叢集
Kafka is a distributed,partitioned,replicated commit logservice。它提供了類似於JMS的特性,但是在設計實現上完全不同,此外它並不是JMS規範的實現。kafka對訊息儲存時根據Topic進行歸類,傳送訊息者成為Producer,訊息接受者成為Consumer,此外kafka叢集有多個kafka例項組成,每個例項(server)成為broker。無論是kafka叢集,還是producer和consumer都依賴於zookeeper來保證系統可用性叢集儲存一些meta資訊。
1.producer:
訊息生產者,釋出訊息到 kafka 叢集的終端或服務。
2.broker:
kafka 叢集中包含的伺服器。
3.topic:
每條釋出到 kafka 叢集的訊息屬於的類別,即 kafka 是面向 topic 的。
4.partition:
partition 是物理上的概念,每個 topic 包含一個或多個 partition。kafka 分配的單位是 partition。
5.consumer:
從 kafka 叢集中消費訊息的終端或服務。
6.Consumer group:
high-level consumer API 中,每個 consumer 都屬於一個 consumer group,每條訊息只能被 consumer group 中的一個 Consumer 消費,但可以被多個 consumer group 消費。
7.replica:
partition 的副本,保障 partition 的高可用。
8.leader:
replica 中的一個角色, producer 和 consumer 只跟 leader 互動。
9.follower:
replica 中的一個角色,從 leader 中複製資料。
10.controller:
kafka 叢集中的其中一個伺服器,用來進行 leader election 以及 各種 failover。
12.zookeeper:
kafka 通過 zookeeper 來儲存叢集的 meta 資訊。
以下是如何從zookeeper中動態讀取kafka叢集中brokers的方法。
public class KafkaBrokerInfoFetcher {
public static void main(String[] args) throws Exception {
ZooKeeper zk = new ZooKeeper("localhost:2181", 10000, null);
List<String> ids = zk.getChildren("/brokers/ids", false);
for (String id : ids) {
String brokerInfo = new String(zk.getData("/brokers/ids/" + id, false, null));
System.out.println(id + ": " + brokerInfo);
}
}
}
結果如下:1: {"jmx_port":-1,"timestamp":"1428512949385","host":"192.168.0.11","version":1,"port":9093}
2: {"jmx_port":-1,"timestamp":"1428512955512","host":"192.168.0.11","version":1,"port":9094}
3: {"jmx_port":-1,"timestamp":"1428512961043","host":"192.168.0.11","version":1,"port":9095}
如果對你有幫助,記得點贊哦~歡迎大家關注我的部落格,可以進群366533258一起交流學習哦~