Kafka部署

1. 簡介 

kafka （官網地址：http://kafka.apache.org）是一款分佈式消息發佈和訂閱的系統，具備高性能和高吞吐率。 

 

i. 消息的發佈（publish）稱做producer，消息的訂閱（subscribe）稱做consumer，中間的存儲陣列稱做broker。

ii. 多個broker協同合做，producer、consumer和broker三者之間經過zookeeper來協調請求和轉發。 

iii. producer產生和推送(push)數據到broker，consumer從broker拉取(pull)數據並進行處理。 

iv. broker端不維護數據的消費狀態，提高了性能。 

v. 直接使用磁盤進行存儲，線性讀寫，速度快：避免了數據在JVM內存和系統內存之間的複製，減小耗性能的建立對象和垃圾回收。 

vi. Kafka使用scala編寫，能夠運行在JVM上。 

2. 安裝： 

a. 首先安裝JRE/JDK 

Linux安裝JDK 

b. 下載kafka 

進入下載頁面：http://kafka.apache.org/downloads.html 

選擇Binary downloads下載 （Source download須要編譯才能使用） 

也能夠直接在linux終端下載： 

Shell代碼 

1. wget -q http://apache.fayea.com/apache-mirror/kafka/0.8.1/kafka_2.8.0-0.8.1.tgz  

c. 解壓 

Shell代碼 

1. tar -xzvf kafka_2.8.0-0.8.1.tgz  
2. rm kafka_2.8.0-0.8.1.tgz  
3. cd kafka_2.8.0-0.8.1  

目錄： 


/bin 啓動和中止命令等。 
/config 配置文件 
/libs 類庫 

d. 修改配置 
Kafka默認開啓JVM壓縮指針，但只是在64位的HotSpot VM受支持，若是安裝了32位的HotSpot VM，須要修改/bin/kafka-run-class.sh文件 

Shell代碼 

1. vi bin/kafka-run-class.sh  

找到以下行： 

Sh代碼 

1. KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true"  

去除-XX:+UseCompressedOops參數 

3. 啓動和中止 

啓動Zookeeper server： 

Shell代碼 

1. bin/zookeeper-server-start.sh config/zookeeper.properties &  

&是爲了能退出命令行 

啓動Kafka server: 

Shell代碼

1. bin/kafka-server-start.sh config/server.properties &  

中止Kafka server 

Shell代碼 

1. bin/kafka-server-stop.sh  

中止Zookeeper server: 

Shell代碼 

1. bin/zookeeper-server-stop.sh  

4. 單機連通性測試 

運行producer： 

Shell代碼 

1. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test  

早版本的Kafka，--broker-list localhost:9092需改成--zookeeper localhost:2181 

運行consumer： 

Shell代碼 

1. bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning  

在producer端輸入字符串並回車，查看consumer端是否顯示。 

5. 分佈式連通性測試 

Zookeeper Server, Kafka Server, Producer都放在服務器server1上，ip地址爲192.168.1.10 
Consumer放在服務器server2上，ip地址爲192.168.1.12。 

分別運行server1的producer和server2的consumer， 

Shell代碼 

1. bin/kafka-console-producer.sh --broker-list 192.168.1.10:9092 --topic test  

 

Shell代碼 

1. bin/kafka-console-consumer.sh --zookeeper 192.168.1.10:2181 --topic test --from-beginning  

在producer的console端輸入字符串，consumer報Connection refused錯誤： 


broker, producer和consumer都註冊到zookeeper上，producer和consumer的參數明確指定。問題出在broker的配置文件server.properties上： 

Properties代碼

1. # Hostname the broker will bind to. If not set, the server will bind to all interfaces  
2. #host.name=localhost  

host名稱沒有指定，就是127.0.0.1，consumer去broker拿數據就有問題。設置爲192.168.1.10，重啓服務就行了。