安裝配置【機器3臺：cdh-yhte01-03】1、下載kafka_2.9.2-0.8.1.1.tgz（http://mirrors.cnnic.cn/apache/kafka/0.8.1.1/kafka_2.9.2-0.8.1.1.tgz）2、 解壓 tar -zxvf kafka_2.9.2-0.8.1.1.tgz3、 確認zookeeper已啟動        [[email protected] ~]# ps -ef | grep zookeeper 4、修改配置檔案 config/server.properties  host.name=cdh-yhte01            log.dirs=/usr/local/kafka/kafka-logs            zookeeper.connect=cdh-yhte01:2181,cdh-yhte02:2181,cdh-yhte03:2181不配置host.name的話，傳送資訊時會出現以下錯誤：kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104)        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87)        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67)        at scala.collection.immutable.Stream.foreach(Stream.scala:526)        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66)        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44)5、啟動kafka            bin/kafka-server-start.sh config/server.properties &6、部署到其他機器[

[email protected] local]# tar -zcvf kafka.tar.gz kafka[[email protected] local]# scp kafka.tar.gz cdh-yhte02:/usr/local                解壓後按如下步驟操作：            修改配置檔案 config/server.properties            【cdh-yhte02】    broker.id=2    port=9092    //埠可以不修改，只有配置在一臺機器的時候需要修改    host.name=cdh-yhte02            【cdh-yhte03】    broker.id=3    port=9093    host.name=cdh-yhte037、啟動02、03機器上的kafka            bin/kafka-server-start.sh config/server.properties &建立一個主題[[email protected] bin]# ./kafka-topics.sh --create --zookeeper cdh-yhte01:2181 --replication-factor 1 --partitions 2 --topic TestKafaCreated topic "TestKafa".        （建立主題後，在另外兩臺機器上都能看到，why？）[[email protected] bin]# ./zookeeper-shell.sh cdh-yhte02:2181Connecting to cdh-yhte02:2181Welcome to ZooKeeper!JLine support is disabledWATCHER::WatchedEvent state:SyncConnected type:None path:nullls /[hbase, zookeeper, admin, consumers, config, controller, brokers, controller_epoch]ls /brokers/topics[TestKafa][[email protected] bin]# ./zookeeper-shell.sh cdh-yhte03:2181Connecting to cdh-yhte03:2181Welcome to ZooKeeper!JLine support is disabledWATCHER::WatchedEvent state:SyncConnected type:None path:nullls /brokers/topics[TestKafa]檢視一個主題[[email protected] bin]# ./kafka-topics.sh --list --zookeeper cdh-yhte01:2181TestKafa檢視詳細資訊[[email protected] bin]# ./kafka-topics.sh --describe --zookeeper cdh-yhte01:2181Topic:GCSJ PartitionCount:3 ReplicationFactor:3 Configs:Topic: GCSJ Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3Topic: GCSJ Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1Topic: GCSJ Partition: 2 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2

• leader：負責處理訊息的讀和寫，leader是從所有節點中隨機選擇的.
• replicas：列出了所有的副本節點，不管節點是否在服務中.
• isr：是正在服務中的節點.

刪除一個主題[[email protected] bin]# ./kafka-topics.sh --list --zookeeper cdh-yhte01:2181TestKafa[[email protected] bin]# ./kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic TestKafa --zookeeper cdh-yhte01:2181deletion succeeded![email protected]:/usr/local/bigdata/kafka_2.11-0.10.0.0/bin# ./kafka-topics.sh --zookeeper node01:2181 --delete --topic CHENLEITopic CHENLEI is marked for deletion.Note: This will have no impact if delete.topic.enable is not set to true.[email protected]:/usr/local/bigdata/kafka_2.11-0.10.0.0/bin# ./kafka-topics.sh --zookeeper node01:2181 --delete --topic CHENLEITopic CHENLEI is already marked for deletion.驗證是否刪除[[email protected] bin]# ./zookeeper-shell.sh cdh-yhte01:2181Connecting to cdh-yhte01:2181Welcome to ZooKeeper!JLine support is disabledWATCHER::WatchedEvent state:SyncConnected type:None path:nullls /[hbase, zookeeper, admin, consumers, config, controller, brokers, controller_epoch]ls /brokers[topics, ids]ls /brokers/topics[]向主題寫入資料./kafka-console-producer.sh --broker-list cdh-yhte01:9091 --sync --topic TestKafa從主題取出資料./kafka-console-consumer.sh --zookeeper node01:2181 --topic CHENLEI --from-beginning檢視消費者位置：bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect cdh-yhte01:2181  --group group1[[email protected] bin]# ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect cdh-yhte01:2181 --group group1Group     Topic     Pid     Offset     logSize     Lag     Ownergroup1     YHTE02 0         41               44         3         none[[email protected] bin]# ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect cdh-yhte01:2181 --group group1Group     Topic     Pid     Offset     logSize     Lag     Ownergroup1 YHTE02     0         44             44             0     group1_leichen-1413446167433-a9b95fe0-0auto.offset.reset：如果zookeeper沒有offset值或offset值超出範圍。那麼就給個初始的offset。有smallest、largest、anything可選，分別表示給當前最小的offset、當前最大的offset、拋異常。預設largest1、資料流將被劃分成獨立的部分（partitions）。這個語義是將劃分的過程留給了producer，producer來指定一個訊息屬於哪個部分。在一個partition中，訊息是按照接受順序排序的，並且會以相同的順序傳送給consumer。這意味著我們需要記錄 每一部分（consumer、topic、and partition）的“high water mark”。 2、在kafka中，high-water mark 以“offset”的形式記錄，這對於各方面來說將變的非常清晰。在kafka中，consumer將保持記錄已經處理的訊息狀態（offset），consumer會將這些記錄資訊寫到zookeeper中，但是如果把這些資訊儲存到consumer正在寫入資料的資料庫會更好，例如：consumer可能會簡單將一些統計資訊寫到OLTP資料庫。這樣 consumer可以把狀態資訊寫到正在進行事務操作的資料庫，這樣解決了分散式一致性問題。很多應用場景下，可以講consumer儲存的狀態資訊（offset）同步到別的地方（如搜尋系統的索引欄位裡，HDFS中），這樣，如果consumer掛了，還能從上次處理的記錄的offset繼續，避免重複處理。另外，這種機制還可以這樣用，直接從頭開始處理，雖然會有出發處理，但是有些場景確實需要這樣做。例如：發現處理過程有BUG，等BUG修復了可以從頭開始處理。