一、Kafka初認識
阿新 • • 發佈：2018-12-14
一、kafka使用背景
1、Kafka使用背景 在我們大量使用分散式資料庫、分散式計算叢集的時候，是否會遇到這樣的一些問題：- 我們想分析下使用者行為（pageviews），以便我們設計出更好的廣告位
- 我想對使用者的搜尋關鍵詞進行統計，分析出當前的流行趨勢
- 有些資料，儲存資料庫浪費，直接儲存硬碟效率又低
舉個例子，生產者消費者，生產者生產雞蛋，消費者消費雞蛋，生產者生產一個雞蛋，消費者就消費一個雞蛋，假設消費者消費雞蛋的時候噎住了（系統宕機了），生產者還在生產雞蛋，那新生產的雞蛋就丟失了。再比如生產者很強勁（大交易量的情況），生產者1秒鐘生產100個雞蛋，消費者1秒鐘只能吃50個雞蛋，那要不了一會，消費者就吃不消了（訊息堵塞，最終導致系統超時），消費者拒絕再吃了，”雞蛋“又丟失了，這個時候我們放個籃子在它們中間，生產出來的雞蛋都放到籃子裡，消費者去籃子裡拿雞蛋，這樣雞蛋就不會丟失了，都在籃子裡，而這個籃子就是”kafka“。雞蛋其實就是“資料流”，系統之間的互動都是通過“資料流”來傳輸的（就是tcp、http什麼的），也稱為報文，也叫“訊息”。訊息佇列滿了，其實就是籃子滿了，”雞蛋“ 放不下了，那趕緊多放幾個籃子，其實就是kafka的擴容。各位現在知道kafka是幹什麼的了吧，它就是那個"籃子"。
定義解釋：
1、Java 和 scala都是執行在JVM上的語言。
2、erlang和最近比較火的和go語言一樣是從程式碼級別就支援高併發的一種語言，所以RabbitMQ天生就有很高的併發效能，但是 有RabbitMQ嚴格按照AMQP進行實現，受到了很多限制。kafka的設計目標是高吞吐量，所以kafka自己設計了一套高效能但是不通用的協議，他也是仿照AMQP（ Advanced Message Queuing Protocol 高階訊息佇列協議）設計的。
3、事物的概念：在資料庫中，多個操作一起提交，要麼操作全部成功，要麼全部失敗。舉個例子， 在轉賬的時候付款和收款，就是一個事物的例子，你給一個人轉賬，你轉成功，並且對方正常行收到款項後，這個操作才算成功，有一方失敗，那麼這個操作就是失敗的。
對應消在息佇列中，就是多條訊息一起傳送，要麼全部成功，要麼全部失敗。3箇中只有ActiveMQ支援，這個是因為，RabbitMQ和Kafka為了更高的效能，而放棄了對事物的支援 。
4、叢集：多臺伺服器組成的整體叫做叢集，這個整體對生產者和消費者來說，是透明的。其實對消費系統組成的叢集新增一臺伺服器減少一臺伺服器對生產者和消費者都是無感之的。
5、負載均衡，對訊息系統來說負載均衡是大量的生產者和消費者向訊息系統發出請求訊息，系統必須均衡這些請求使得每一臺伺服器的請求達到平衡，而不是大量的請求，落到某一臺或幾臺，使得這幾臺伺服器高負荷或超負荷工作，嚴重情況下會停止服務或宕機。
6、動態擴容是很多公司要求的技術之一，不支援動態擴容就意味著停止服務，這對很多公司來說是不可以接受的。
注：
阿里巴巴的Metal,RocketMQ都有Kafka的影子，他們要麼改造了Kafka或者借鑑了Kafka，最後Kafka的動態擴容是通過Zookeeper來實現的。
Zookeeper是一種在分散式系統中被廣泛用來作為：分散式狀態管理、分散式協調管理、分散式配置管理、和分散式鎖服務的叢集。kafka增加和減少伺服器都會在Zookeeper節點上觸發相應的事件kafka系統會捕獲這些事件，進行新一輪的負載均衡，客戶端也會捕獲這些事件來進行新一輪的處理。
二、kafka相關概念
1、 AMQP協議
Advanced Message Queuing Protocol （高階訊息佇列協議） The Advanced Message Queuing Protocol (AMQP)：是一個標準開放的應用層的訊息中介軟體（Message Oriented Middleware）協議。AMQP定義了通過網路傳送的位元組流的資料格式。因此相容性非常好，任何實現AMQP協議的程式都可以和與AMQP協議相容的其他程式互動，可以很容易做到跨語言，跨平臺。 上面說的3種比較流行的訊息佇列協議，要麼支援AMQP協議，要麼借鑑了AMQP協議的思想進行了開發、實現、設計。 2、 一些基本的概念
3、Kafka架構
生產者生產訊息、kafka叢集、消費者獲取訊息這樣一種架構，如下圖：
kafka叢集中的訊息，是通過Topic（主題）來進行組織的，如下圖：
一些基本的概念：
1、主題（Topic）：一個主題類似新聞中的體育、娛樂、教育等分類概念，在實際工程中通常一個業務一個主題。
2、分割槽（Partition）：一個Topic中的訊息資料按照多個分割槽組織，分割槽是kafka訊息佇列組織的最小單位，一個分割槽可以看作是一個FIFO（ First Input First Output的縮寫，先入先出佇列）的佇列。
kafka分割槽是提高kafka效能的關鍵所在，當你發現你的叢集效能不高時，常用手段就是增加Topic的分割槽，分割槽裡面的訊息是按照從新到老的順序進行組織，消費者從佇列頭訂閱訊息，生產者從佇列尾新增訊息。
工作圖：
備份（Replication）：為了保證分散式可靠性，kafka0.8開始對每個分割槽的資料進行備份（不同的Broker上），防止其中一個Broker宕機造成分割槽上的資料不可用。
kafka0.7是一個很大的改變：1、增加了備份2、增加了控制借點概念，增加了叢集領導者選舉 。
三、zookeeper叢集搭建
Kafka叢集是把狀態儲存在Zookeeper中的，首先要搭建Zookeeper叢集。 1、軟體環境 （3臺伺服器-我的測試） 192.168.7.100 server1 192.168.7.101 server2 192.168.7.107 server3 1、Linux伺服器一臺、三臺、五臺、（2*n+1），Zookeeper叢集的工作是超過半數才能對外提供服務，3臺中超過兩臺超過半數，允許1臺掛掉 ，是否可以用偶數，其實沒必要。 如果有四臺那麼掛掉一臺還剩下三臺伺服器，如果在掛掉一個就不行了，這裡記住是超過半數。 2、Java jdk1.7 zookeeper是用java寫的所以他的需要JAVA環境，java是執行在java虛擬機器上的 3、Zookeeper的穩定版本Zookeeper 3.4.6版本 2、配置&安裝Zookeeper 下面的操作是：3臺伺服器統一操作 1、安裝Javayum list java* yum -y install java-1.7.0-openjdk*
2、下載Zookeeper
首先要注意在生產環境中目錄結構要定義好，防止在專案過多的時候找不到所需的專案
#我的目錄統一放在/opt下面 #首先建立Zookeeper專案目錄 mkdir zookeeper #專案目錄 mkdir zkdata #存放快照日誌 mkdir zkdatalog#存放事物日誌
下載Zookeeper
#下載軟體 cd /opt/zookeeper/ wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz #解壓軟體 tar -zxvf zookeeper-3.4.6.tar.gz
3、修改配置檔案
進入到解壓好的目錄裡面的conf目錄中，檢視#進入conf目錄 /opt/zookeeper/zookeeper-3.4.6/conf #檢視 [[email protected]]$ ll -rw-rw-r--. 1 1000 1000 535 Feb 20 2014 configuration.xsl -rw-rw-r--. 1 1000 1000 2161 Feb 20 2014 log4j.properties -rw-rw-r--. 1 1000 1000 922 Feb 20 2014 zoo_sample.cfg
#zoo_sample.cfg 這個檔案是官方給我們的zookeeper的樣板檔案，給他複製一份命名為zoo.cfg，zoo.cfg是官方指定的檔案命名規則。
3臺伺服器的配置檔案
tickTime=2000 initLimit=10 syncLimit=5 dataDir=/opt/zookeeper/zkdata dataLogDir=/opt/zookeeper/zkdatalog clientPort=12181 server.1=192.168.7.100:12888:13888 server.2=192.168.7.101:12888:13888 server.3=192.168.7.107:12888:13888 #server.1 這個1是伺服器的標識也可以是其他的數字， 表示這個是第幾號伺服器，用來標識伺服器，這個標識要寫到快照目錄下面myid檔案裡 #192.168.7.107為叢集裡的IP地址，第一個埠是master和slave之間的通訊埠，預設是2888，第二個埠是leader選舉的埠，叢集剛啟動的時候選舉或者leader掛掉之後進行新的選舉的埠預設是3888
配置檔案解釋：
#tickTime： 這個時間是作為 Zookeeper 伺服器之間或客戶端與伺服器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。 #initLimit： 這個配置項是用來配置 Zookeeper 接受客戶端（這裡所說的客戶端不是使用者連線 Zookeeper 伺服器的客戶端，而是 Zookeeper 伺服器叢集中連線到 Leader 的 Follower 伺服器）初始化連線時最長能忍受多少個心跳時間間隔數。當已經超過 5個心跳的時間（也就是 tickTime）長度後 Zookeeper 伺服器還沒有收到客戶端的返回資訊，那麼表明這個客戶端連線失敗。總的時間長度就是 5*2000=10 秒 #syncLimit： 這個配置項標識 Leader 與Follower 之間傳送訊息，請求和應答時間長度，最長不能超過多少個 tickTime 的時間長度，總的時間長度就是5*2000=10秒 #dataDir： 快照日誌的儲存路徑 #dataLogDir： 事物日誌的儲存路徑，如果不配置這個那麼事物日誌會預設儲存到dataDir制定的目錄，這樣會嚴重影響zk的效能，當zk吞吐量較大的時候，產生的事物日誌、快照日誌太多 #clientPort： 這個埠就是客戶端連線 Zookeeper 伺服器的埠，Zookeeper 會監聽這個埠，接受客戶端的訪問請求。修改他的埠改大點
建立myid檔案
#server1 echo "1" > /opt/zookeeper/zkdata/myid #server2 echo "2" > /opt/zookeeper/zkdata/myid #server3 echo "3" > /opt/zookeeper/zkdata/myid
4、重要配置說明
1、myid檔案和server.myid 在快照目錄下存放的標識本臺伺服器的檔案，他是整個zk叢集用來發現彼此的一個重要標識。
2、zoo.cfg 檔案是zookeeper配置檔案 在conf目錄裡。
3、log4j.properties檔案是zk的日誌輸出檔案 在conf目錄裡用java寫的程式基本上有個共同點日誌都用log4j，來進行管理。
# Define some default values that can be overridden by system properties zookeeper.root.logger=INFO, CONSOLE #日誌級別 zookeeper.console.threshold=INFO #使用下面的console來列印日誌 zookeeper.log.dir=. #日誌列印到那裡，是咱們啟動zookeeper的目錄 （建議設定統一的日誌目錄路徑） zookeeper.log.file=zookeeper.log zookeeper.log.threshold=DEBUG zookeeper.tracelog.dir=. zookeeper.tracelog.file=zookeeper_trace.log # # ZooKeeper Logging Configuration # # Format is "<default threshold> (, <appender>)+ # DEFAULT: console appender only log4j.rootLogger=${zookeeper.root.logger} # Example with rolling log file #log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE # Example with rolling log file and tracing #log4j.rootLogger=TRACE, CONSOLE, ROLLINGFILE, TRACEFILE # # Log INFO level and above messages to the console # log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender log4j.appender.CONSOLE.Threshold=${zookeeper.console.threshold} log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n # Add ROLLINGFILE to rootLogger to get log file output # Log DEBUG level and above messages to a log file log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold} log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file} # Max log file size of 10MB log4j.appender.ROLLINGFILE.MaxFileSize=10MB # uncomment the next line to limit number of backup files #log4j.appender.ROLLINGFILE.MaxBackupIndex=10 log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n # # Add TRACEFILE to rootLogger to get log file output # Log DEBUG level and above messages to a log file log4j.appender.TRACEFILE=org.apache.log4j.FileAppender log4j.appender.TRACEFILE.Threshold=TRACE log4j.appender.TRACEFILE.File=${zookeeper.tracelog.dir}/${zookeeper.tracelog.file} log4j.appender.TRACEFILE.layout=org.apache.log4j.PatternLayout ### Notice we are including log4j's NDC here (%x) log4j.appender.TRACEFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L][%x] - %m%n configuration for log4j
4、zkEnv.sh和zkServer.sh檔案
zkServer.sh 主的管理程式檔案 zkEnv.sh 是主要配置，zookeeper叢集啟動時配置環境變數的檔案 5、還有一個需要注意 ZooKeeper server will not remove old snapshots and log files when using the default configuration (see autopurge below), this is the responsibility of the operator zookeeper不會主動的清除舊的快照和日誌檔案，這個是操作者的責任。但是可以通過命令去定期的清理。
#!/bin/bash #snapshot file dir dataDir=/opt/zookeeper/zkdata/version-2 #tran log dir dataLogDir=/opt/zookeeper/zkdatalog/version-2 #Leave 66 files count=66 count=$[$count+1] ls -t $dataLogDir/log.* | tail -n +$count | xargs rm -f ls -t $dataDir/snapshot.* | tail -n +$count | xargs rm -f #以上這個指令碼定義了刪除對應兩個目錄中的檔案，保留最新的66個檔案，可以將他寫到crontab中，設定為每天凌晨2點執行一次就可以了。 #zk log dir del the zookeeper log #logDir= #ls -t $logDir/zookeeper.log.* | tail -n +$count | xargs rm -f
其他方法：
第二種：使用ZK的工具類PurgeTxnLog，它的實現了一種簡單的歷史檔案清理策略，可以在這裡看一下他的使用方法 http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html
第三種：對於上面這個執行，ZK自己已經寫好了指令碼，在bin/zkCleanup.sh中，所以直接使用這個指令碼也是可以執行清理工作的。
第四種：從3.4.0開始，zookeeper提供了自動清理snapshot和事務日誌的功能，通過配置 autopurge.snapRetainCount 和 autopurge.purgeInterval 這兩個引數能夠實現定時清理了。這兩個引數都是在zoo.cfg中配置的：
autopurge.purgeInterval 這個引數指定了清理頻率，單位是小時，需要填寫一個1或更大的整數，預設是0，表示不開啟自己清理功能。 autopurge.snapRetainCount 這個引數和上面的引數搭配使用，這個引數指定了需要保留的檔案數目。預設是保留3個。 推薦使用第一種方法，對於運維人員來說，將日誌清理工作獨立出來，便於統一管理也更可控。畢竟zk自帶的一些工具並不怎麼給力。 5、啟動服務並檢視 1、啟動服務#進入到Zookeeper的bin目錄下 cd /opt/zookeeper/zookeeper-3.4.6/bin #啟動服務（3臺都需要操作） ./zkServer.sh start
2、檢查服務狀態
#檢查伺服器狀態 ./zkServer.sh status
通過status就能看到狀態：
./zkServer.sh status JMX enabled by default Using config: /opt/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg #配置檔案 Mode: follower #他是否為領導
zk叢集一般只有一個leader，多個follower，主一般是相應客戶端的讀寫請求，而從主同步資料，當主掛掉之後就會從follower裡投票選舉一個leader出來。
可以用“jps”檢視zk的程序，這個是zk的整個工程的main
#執行命令jps 20348 Jps 4233 QuorumPeerMain
四、kafka叢集搭建
1、軟體環境 1、linux一臺或多臺，大於等於2 2、已經搭建好的zookeeper叢集 3、軟體版本kafka_2.11-0.9.0.1.tgz 2、建立目錄並下載安裝軟體#建立目錄 cd /opt/ mkdir kafka #建立專案目錄 cd kafka mkdir kafkalogs #建立kafka訊息目錄，主要存放kafka訊息 #下載軟體 wget http://apache.opencas.org/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz #解壓軟體 tar -zxvf kafka_2.11-0.9.0.1.tgz
3、修改配置檔案
進入到config目錄cd /opt/kafka/kafka_2.11-0.9.0.1/config/
主要關注：server.properties 這個檔案即可，我們可以發現在目錄下：
有很多檔案，這裡可以發現有Zookeeper檔案，我們可以根據Kafka內帶的zk叢集來啟動，但是建議使用獨立的zk叢集
-rw-r--r--. 1 root root 5699 Feb 22 09:41 192.168.7.101 -rw-r--r--. 1 root root 906 Feb 12 08:37 connect-console-sink.properties -rw-r--r--. 1 root root 909 Feb 12 08:37 connect-console-source.properties -rw-r--r--. 1 root root 2110 Feb 12 08:37 connect-distributed.properties -rw-r--r--. 1 root root 922 Feb 12 08:38 connect-file-sink.properties -rw-r--r--. 1 root root 920 Feb 12 08:38 connect-file-source.properties -rw-r--r--. 1 root root 1074 Feb 12 08:37 connect-log4j.properties -rw-r--r--. 1 root root 2055 Feb 12 08:37 connect-standalone.properties -rw-r--r--. 1 root root 1199 Feb 12 08:37 consumer.properties -rw-r--r--. 1 root root 4369 Feb 12 08:37 log4j.properties -rw-r--r--. 1 root root 2228 Feb 12 08:38 producer.properties -rw-r--r--. 1 root root 5699 Feb 15 18:10 server.properties -rw-r--r--. 1 root root 3325 Feb 12 08:37 test-log4j.properties -rw-r--r--. 1 root root 1032 Feb 12 08:37 tools-log4j.properties -rw-r--r--. 1 root root 1023 Feb 12 08:37 zookeeper.properties
修改配置檔案：
broker.id=0 #當前機器在叢集中的唯一標識，和zookeeper的myid性質一樣 port=19092 #當前kafka對外提供服務的埠預設是9092 host.name=192.168.7.100 #這個引數預設是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。 num.network.threads=3 #這個是borker進行網路處理的執行緒數 num.io.threads=8 #這個是borker進行I/O處理的執行緒數 log.dirs=/opt/kafka/kafkalogs/ #訊息存放的目錄，這個目錄可以配置為“，”逗號分割的表示式，上面的num.io.threads要大於這個目錄的個數這個目錄，如果配置多個目錄，新建立的topic他把訊息持久化的地方是，當前以逗號分割的目錄中，那個分割槽數最少就放那一個 socket.send.buffer.bytes=102400 #傳送緩衝區buffer大小，資料不是一下子就傳送的，先回儲存到緩衝區了到達一定的大小後在傳送，能提高效能 socket.receive.buffer.bytes=102400 #kafka接收緩衝區大小，當資料到達一定大小後在序列化到磁碟 socket.request.max.bytes=104857600 #這個引數是向kafka請求訊息或者向kafka傳送訊息的請請求的最大數，這個值不能超過java的堆疊大小 num.partitions=1 #預設的分割槽數，一個topic預設1個分割槽數 log.retention.hours=168 #預設訊息的最大持久化時間，168小時，7天 message.max.byte=5242880 #訊息儲存的最大值5M default.replication.factor=2 #kafka儲存訊息的副本數，如果一個副本失效了，另一個還可以繼續提供服務 replica.fetch.max.bytes=5242880 #取訊息的最大直接數 log.segment.bytes=1073741824 #這個引數是：因為kafka的訊息是以追加的形式落地到檔案，當超過這個值的時候，kafka會新起一個檔案 log.retention.check.interval.ms=300000 #每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄檢視是否有過期的訊息如果有，刪除 log.cleaner.enable=false #是否啟用log壓縮，一般不用啟用，啟用的話可以提高效能 zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:1218 #設定zookeeper的連線埠
上面是引數的解釋，實際的修改項為：
#broker.id=0 每臺伺服器的broker.id都不能相同 #hostname host.name=192.168.7.100 #在log.retention.hours=168 下面新增下面三項 message.max.byte=5242880 default.replication.factor=2 replica.fetch.max.bytes=5242880 #設定zookeeper的連線埠 zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:12181
4、啟動Kafka叢集並測試
1、啟動服務
#從後臺啟動Kafka叢集（3臺都需要啟動） cd
/opt/kafka/kafka_2.11-0.9.0.1//bin #進入到kafka的bin目錄 ./kafka-server-start.sh -daemon ../config/server.properties
2、檢查服務是否啟動
#執行命令jps 20348 Jps 4233 QuorumPeerMain 18991 Kafka
3、建立Topic來驗證是否建立成功
更多請看官方文件：http://kafka.apache.org/documentation.html
#建立Topic ./kafka-topics.sh --create --zookeeper 192.168.7.100:12181 --replication-factor 2 --partitions 1 --topic shuaige #解釋 --replication-factor 2 #複製兩份 --partitions 1 #建立1個分割槽 --topic #主題為shuaige '''在一臺伺服器上建立一個釋出者''' #建立一個broker，釋出者 ./kafka-console-producer.sh --broker-list 192.168.7.100:19092 --topic shuaige '''在一臺伺服器上建立一個訂閱者''' ./kafka-console-consumer.sh --zookeeper localhost:12181 --topic shuaige --from-beginning
測試（在釋出者那裡釋出訊息看看訂閱者那裡是否能正常收到~）：
4、其他命令
大部分命令可以去官方文件檢視
4.1、檢視topic
./kafka-topics.sh --list --zookeeper localhost:12181 #就會顯示我們建立的所有topic
4.2、檢視topic狀態
/kafka-topics.sh --describe --zookeeper localhost:12181 --topic shuaige
#下面是顯示資訊
Topic:ssports PartitionCount:1 ReplicationFactor:2 Configs:
Topic: shuaige Partition: 0 Leader: 1 Replicas: 0,1 Isr: 1
#分割槽為為1 複製因子為2 他的 shuaige的分割槽為0
#Replicas: 0,1 複製的為0，1
#
OKkafka叢集搭建完畢
5、其他說明標註
5.1、日誌說明
預設kafka的日誌是儲存在/opt/kafka/kafka_2.10-0.9.0.0/logs目錄下的，這裡說幾個需要注意的日誌
server.log #kafka的執行日誌 state-change.log #kafka他是用zookeeper來儲存狀態，所以他可能會進行切換，切換的日誌就儲存在這裡 controller.log #kafka選擇一個節點作為“controller”,當發現有節點down掉的時候它負責在游泳分割槽的所有節點中選擇新的leader,這使得Kafka可以批量的高效的管理所有分割槽節點的主從關係。如果controller down掉了，活著的節點中的一個會備切換為新的controller.
5.2、上面的大家你完成之後可以登入zk來檢視zk的目錄情況
#使用客戶端進入zk ./zkCli.sh -server 127.0.0.1:12181 #預設是不用加’-server‘引數的因為我們修改了他的埠 #檢視目錄情況 執行“ls /” [zk: 127.0.0.1:12181(CONNECTED) 0] ls / #顯示結果：[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch] ''' 上面的顯示結果中：只有zookeeper是，zookeeper原生的，其他都是Kafka建立的 ''' #標註一個重要的 [zk: 127.0.0.1:12181(CONNECTED) 1] get /brokers/ids/0 {"jmx_port":-1,"timestamp":"1456125963355","endpoints":["PLAINTEXT://192.168.7.100:19092"],"host":"192.168.7.100","version":2,"port":19092} cZxid = 0x1000001c1 ctime = Mon Feb 22 15:26:03 CST 2016 mZxid = 0x1000001c1 mtime = Mon Feb 22 15:26:03 CST 2016 pZxid = 0x1000001c1 cversion = 0 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x152e40aead20016 dataLength = 139 numChildren = 0 [zk: 127.0.0.1:12181(CONNECTED) 2] #還有一個是檢視partion [zk: 127.0.0.1:12181(CONNECTED) 7] get /brokers/topics/shuaige/partitions/0 null cZxid = 0x100000029 ctime = Mon Feb 22 10:05:11 CST 2016 mZxid = 0x100000029 mtime = Mon Feb 22 10:05:11 CST 2016 pZxid = 0x10000002a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 0 numChildren = 1 [zk: 127.0.0.1:12181(CONNECTED) 8]
參考資料說明：
《極客學院kafka學習視訊》