初識中間件Kafka

初識中間件Kafka

Author：SimplelWu

什麼是消息中間件？

• 非底層操做系統軟件，非業務應用軟件，不是直接給最終用戶使用的，不能直接給客戶帶來價值的軟件統稱爲中間件
• 關注於數據的發送和接收，利用高效可靠的異步消息傳遞機制集成分佈式系統。

什麼是Kafka？

Kafka是一種高吞吐量的分佈式發佈訂閱消息系統，是一個分佈式的、分區的、可靠的分佈式日誌存儲服務。它經過一種獨一無二的設計提供了一個消息系統的功能。

kafka官方：http://kafka.apache.org/

Kafka做爲一個分佈式的流平臺，這到底意味着什麼？

咱們認爲，一個流處理平臺具備三個關鍵能力：

• 發佈和訂閱消息（流），在這方面，它相似於一個消息隊列或企業消息系統。
• 以容錯的方式存儲消息（流）。
• 在消息流發生時處理它們。

什麼是kakfa的優點?

它應用於2大類應用：

• 構建實時的流數據管道，可靠地獲取系統和應用程序之間的數據。
• 構建實時流的應用程序，對數據流進行轉換或反應。

kafka有四個核心API

• 應用程序使用 Producer API 發佈消息到1個或多個topic（主題）。
• 應用程序使用 Consumer API 來訂閱一個或多個topic，並處理產生的消息。
• 應用程序使用 Streams API 充當一個流處理器，從1個或多個topic消費輸入流，並生產一個輸出流到1個或多個輸出topic，有效地將輸入流轉換到輸出流。
• Connector API容許構建或運行可重複使用的生產者或消費者，將topic鏈接到現有的應用程序或數據系統。例如，一個關係數據庫的鏈接器可捕獲每個變化。

Client和Server之間的通信，是經過一條簡單、高性能而且和開發語言無關的TCP協議。而且該協議保持與老版本的兼容。Kafka提供了Java Client（客戶端）。除了Java Client外，還有很是多的其它編程語言的Client。

主流消息中間件比較

	ActiveMQ	RabbitMQ	Kafka
跨語言	支持（Java優先）	語言無關	支持（Java優先）
支持協議	OpenWire，Stomp， XMPP，AMQP	AMQP
優勢	遵循JMS規範，安裝部署方便。	繼承Erlang天生的併發性，最初用於金融行業，穩定性，安全性有保障。	依賴zk，可動態擴展節點，高性能，高吞吐量，無線擴容消息可指定追溯。
缺點	根據其餘用戶反饋，會莫名丟失消息，目前重心在下一代的apolle上，目前社區不活躍，對5.X維護較少。	Erlang語言難度較大，不支持動態擴展。	嚴格的順序機制，不支持消息優先級，不支持標準的消息協議，不利於平臺遷移。
綜合評價	適合中小企業消息應用場景，不適合上千個隊列的應用場景。	適合對穩定性要求較高的企業應用。	通常應用在大數據日誌處理或對實時性，可靠性要求稍低的場景。

Kafka好處

• 可靠性 - Kafka是分佈式，分區，複製和容錯的。
• 可擴展性 - Kafka消息傳遞系統輕鬆縮放，無需停機。
• 耐用性 - Kafka使用分佈式提交日誌，這意味着消息會盡量快地保留在磁盤上，所以它是持久的。
• 性能 - Kafka對於發佈和訂閱消息都具備高吞吐量。 即便存儲了許多TB的消息，它也保持穩定的性能。

Kafka很是快，並保證零停機和零數據丟失。

應用場景

• 指標 - Kafka一般用於操做監控數據。 這涉及聚合來自分佈式應用程序的統計信息，以產生操做數據的集中饋送。
• 日誌聚合解決方案 - Kafka可用於跨組織從多個服務收集日誌，並使它們以標準格式提供給多個服務器。
• 流處理 - 流行的框架(如Storm和Spark Streaming)從主題中讀取數據，對其進行處理，並將處理後的數據寫入新主題，供用戶和應用程序使用。 Kafka的強耐久性在流處理的上下文中也很是有用。

Kafka相關術語

序號	組件和說明
1	Topics（主題）屬於特定類別的消息流稱爲主題。 數據存儲在主題中。主題被拆分紅分區。 對於每一個主題，Kafka保存一個分區的數據。 每一個這樣的分區包含不可變有序序列的消息。 分區被實現爲具備相等大小的一組分段文件。
2	Partition（分區）主題可能有許多分區，所以它能夠處理任意數量的數據。
3	Partition offset（分區偏移）每一個分區消息具備稱爲 offset 的惟一序列標識。
4	Replicas of partition（分區備份）副本只是一個分區的備份。 副本從不讀取或寫入數據。 它們用於防止數據丟失。
5	Brokers（經紀人）代理是負責維護髮布數據的簡單系統。 每一個代理中的每一個主題能夠具備零個或多個分區。 假設，若是在一個主題和N個代理中有N個分區，每一個代理將有一個分區。假設在一個主題中有N個分區而且多於N個代理(n + m)，則第一個N代理將具備一個分區，而且下一個M代理將不具備用於該特定主題的任何分區。假設在一個主題中有N個分區而且小於N個代理(n-m)，每一個代理將在它們之間具備一個或多個分區共享。 因爲代理之間的負載分佈不相等，不推薦使用此方案。
6	Kafka Cluster（Kafka集羣）Kafka有多個代理被稱爲Kafka集羣。 能夠擴展Kafka集羣，無需停機。 這些集羣用於管理消息數據的持久性和複製。
7	Producers（生產者）生產者是發送給一個或多個Kafka主題的消息的發佈者。 生產者向Kafka經紀人發送數據。 每當生產者將消息發佈給代理時，代理只需將消息附加到最後一個段文件。 實際上，該消息將被附加到分區。 生產者還能夠向他們選擇的分區發送消息。
8	Consumers（消費者）Consumers從經紀人處讀取數據。 消費者訂閱一個或多個主題，並經過從代理中提取數據來使用已發佈的消息。
9	Leader（領導者） Leader 是負責給定分區的全部讀取和寫入的節點。每一個分區都有一個服務器充當Leader 。
10	Follower（追隨者）跟隨領導者指令的節點被稱爲Follower。 若是領導失敗，一個追隨者將自動成爲新的領導者。 跟隨者做爲正常消費者，拉取消息並更新其本身的數據存儲。

使用Kafka

1. 安裝jdk
2. 安裝zookepper 官方：http://zookeeper.apache.org/
3. 安裝kafka

我這裏jdk是已經安裝好的。

安裝zookepper：

tar -zxvf zookeeper-3.4.13.tar.gz #解壓
cd zookeeper-3.4.13/config #進入配置目錄
#zookeeper運行須要config裏有config文件。可是解壓後默認只有zoo_sample.cfg，咱們將名字修改下便可
mv zoo_sample.cfg zoo.cfg #修改配置文件名字

啓動zookeper，來到bin目錄：

./zkServer.sh start #啓動zookepper

中止zookeper，來到bin目錄：

./zkServer.sh start #中止zookepper

kafka下載：https://www.apache.org/dyn/closer.cgi?path=/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz

使用Kafka，

tar -zxvf kafka_2.11-2.1.0.tgz #解壓kafka

啓動zookpper服務,來到kafka的bin目錄：

./zookeeper-server-start.sh config/zookeeper.properties #啓動服務

啓動kafka服務：

./kafka-server-start.sh config/server.properties #啓動kafka服務

建立一個主題:

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test #topic_name

topic_name：主題的名字'test'。

建立好後查看主題:

kafka-topics.sh --list --zookeeper localhost:2181

Kafka提供了一個命令行的工具，能夠從輸入文件或者命令行中讀取消息併發送給Kafka集羣。每一行是一條消息。
運行producer（生產者）,而後在控制檯輸入幾條消息到服務器。

發送消息：

./kafka-console-producer.sh --broker-list localhost:9092 --topic test #主題爲test

進入以後就可發送消息！！！

Kafka也提供了一個消費消息的命令行工具，將存儲的信息輸出出來。

消費消息：

#topic主題須要與被消費的主題對應上
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

Kafka經常使用命令

#查看全部主題列表
kafka-topics.sh --zookeeper localhost:2181 --list
#查看指定topic信息
kafka-topics.sh --zookeeper localhost:2181 --describe --topic topic_name
#控制檯向topic生產數據
kafka-console-producer.sh --broker-list localhost:9092 --topic topic_name 
#控制檯消費topic的數據
kafka-console-consumer.sh  --zookeeper localhost:2181  --topic topic_name --from-beginning 
#查看topic某分區偏移量最大（小）值
kafka-run-class.sh kafka.tools.GetOffsetShell --topic hive-mdatabase-hostsltable  --time -1 --broker-list localhost:9092 --partitions 0 
#增長topic分區數
kafka-topics.sh --zookeeper localhost:2181  --alter --topic topic_name --partitions 10
#刪除topic，慎用，只會刪除zookeeper中的元數據，消息文件須手動刪除
kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper localhost:2181 --topic topic_name