Kafka安裝（極簡版）

Kafka簡介

kafka是一個高吞吐量的分佈式消息隊列，具備高性能、持久化、多副本備份、橫向擴展能力，一般用於大數據及流處理平臺。消息隊列裏都有生產者/消費者的概念，生產者往隊列裏寫消息，而消費者則是從隊列裏獲取消息。通常在架構設計中起到解耦、削峯、異步處理的做用。

kafka對外使用topic的概念，生產者往topic裏寫消息，消費者則從topic裏讀消息。爲了作到水平擴展，一個topic實際是由多個partition組成的，遇到瓶頸時，能夠經過增長partition的數量來進行橫向擴容。在單個parition內是保證消息有序。每新寫一條消息，kafka就是在對應的文件append寫，因此性能很是高。

kafka的整體數據流是這樣的：

大概用法就是，Producers往Brokers裏面的指定Topic中寫消息，Consumers從Brokers裏面拉去指定Topic的消息，而後進行業務處理。

圖中有兩個topic，topic 0有兩個partition，topic 1有一個partition，三副本備份。能夠看到consumer gourp 1中的consumer 2沒有分到partition處理，這是有可能出現的。

kafka須要依賴zookeeper存儲一些元信息，而kafka也自帶了zookeeper。其中broker、topics、partitions的一些元信息用zookeeper來存儲，監控和路由啥的也都會用到zookeeper。

kafka名詞解釋：

• producer：生產者。
• consumer：消費者。
• topic: 消息以topic爲類別記錄，Kafka將消息種子(Feed)分門別類，每一類的消息稱之爲一個主題(Topic)。
• partitions：每一個Topics劃分爲一個或者多個partition，而且partition中的每條消息都被標記了一個sequential id ，也就是offset，而且存儲的數據是可配置存儲時間的
• broker：以集羣的方式運行，能夠由一個或多個服務組成，每一個服務叫作一個broker，消費者能夠訂閱一個或多個主題(topic)，並從Broker拉數據，從而消費這些已發佈的消息。

在kafka中每一個消息（也叫做record記錄，也被稱爲消息）一般是由一個key，一個value和時間戳構成。

kafka有四個核心的API：

• 應用程序使用Producer API發佈消息到1個或多個topic中。
• 應用程序使用Consumer API來訂閱一個或多個topic，並處理產生的消息。
• 應用程序使用Streams API充當一個流處理器，從1個或多個topic消費輸入流,併產生一個輸出流到1個或多個topic，有效地將輸入流轉換到輸出流。
• Connector API容許構建或運行可重複使用的生產者或消費者，將topic連接到現有的應用程序或數據系統。

kafka就先介紹到這，網絡上有不少相關的理論文章，因此這裏不過多贅述了，也能夠直接查看官方文檔。官方文檔地址以下：

http://kafka.apache.org/intro.html

---

單實例安裝

本小節咱們來在CentOS7上安裝Kafka，因爲kafka是由Scala和Java語言編寫的，因此前提須要準備好java運行環境，我這裏java環境是1.8的，因爲jdk的安裝配置都比較簡單，這裏就不演示jdk的安裝過程了，直接安裝Kafka。

到官網上覆制下載地址，使用wget命令進行下載並解壓：

[root@study-01 ~]# cd /usr/local/src/
[root@study-01 /usr/local/src]# wget http://mirrors.hust.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz
[root@study-01 /usr/local/src]# tar -zxvf kafka_2.11-2.0.0.tgz
[root@study-01 /usr/local/src]# mv kafka_2.11-2.0.0 /usr/local/kafka
[root@study-01 /usr/local/src]# cd !$

沒有特殊要求的話，咱們使用默認的kafka配置便可。若你但願kafka可以被外部機器訪問，則須要配置一下你機器的外網ip地址及Kafka監聽端口，另外咱們可能還須要配置broker的id和kafka日誌文件的存儲目錄，以下：

[root@study-01 /usr/local/kafka]# vim ./config/server.properties
listeners=PLAINTEXT://192.168.190.129:9092
advertised.listeners=PLAINTEXT://192.168.190.129:9092

broker.id=1  # broker的id，必須是集羣中惟一的
log.dirs=/tmp/kafka-logs   # kafka日誌文件的存儲目錄
[root@study-01 /usr/local/kafka]#

如今咱們就可使用kafka了，因爲kafka依賴zookeeper，因此咱們在啓動kafka前須要先啓動kafka自帶的zookeeper服務：

[root@study-01 /usr/local/kafka]# nohup ./bin/zookeeper-server-start.sh ./config/zookeeper.properties > zookeeper.out &

zookeeper服務啓動成功後，啓動kafka：

[root@study-01 /usr/local/kafka]# nohup ./bin/kafka-server-start.sh ./config/server.properties > kafka.out &

兩個服務都啓動成功後，監聽的端口以下：

[root@study-01 ~]# netstat -lntp |grep java
tcp6       0      0 :::38031                :::*                    LISTEN      3629/java
tcp6       0      0 :::33620                :::*                    LISTEN      3945/java
tcp6       0      0 :::9092                 :::*                    LISTEN      3945/java
tcp6       0      0 :::2181                 :::*                    LISTEN      3629/java
[root@study-01 ~]#

---

接下來咱們測試一下kafka是否正常可用，首先建立一個topic，命令以下：

[root@study-01 /usr/local/kafka]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic hello
Created topic "hello".
[root@study-01 /usr/local/kafka]#

建立topic成功後，使用以下命令測試是否能正常獲取topic列表：

[root@study-01 /usr/local/kafka]# ./bin/kafka-topics.sh --list --zookeeper localhost:2181
hello
[root@study-01 /usr/local/kafka]#

如今能夠肯定topic建立成功了，而後咱們來啓動producer，測試往一個topic上發送消息：

[root@study-01 /usr/local/kafka]# ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic hello>hello world
>hello kafka 
>

接着啓動consumer，測試從一個topic上消費消息：

[root@study-01 /usr/local/kafka]# ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello --from-beginning
hello world
hello kafka

經過以上測試，能夠看到，kafka可以正常的建立topic進行發送/接收消息，那麼就表明咱們安裝成功了。