初識中間件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