今天給公司小夥伴培訓的 Kafka 入門，瞭解一下？

時間 2019-11-06

標籤今天公司夥伴培訓 kafka 入門瞭解一下欄目應用數學简体版

原文原文鏈接

消息中間件/系統

中間件

中間件是一類鏈接軟件組件和應用的計算機軟件，它包括一組服務。以便於運行在一臺或多臺機器上的多個軟件經過網絡進行交互。java

該技術所提供的互操做性，推進了一致分佈式體系架構的演進，該架構一般用於支持並簡化那些複雜的分佈式應用程序，它包括 web服務器、事務監控器和消息隊列軟件。linux

消息隊列

在計算機科學中，消息隊列（英語：Message queue）是一種進程間通訊或同一進程的不一樣線程間的通訊方式，軟件的貯列用來處理一系列的輸入，一般是來自用戶。web

消息隊列提供了異步的通訊協議，每個貯列中的紀錄包含詳細說明的數據，包含發生的時間，輸入設備的種類，以及特定的輸入參數，也就是說：消息的發送者和接收者不須要同時與消息隊列交互。消息會保存在隊列中，直到接收者取回它。算法

消息隊列經常保存在鏈表結構中。擁有權限的進程能夠向消息隊列中寫入或讀取消息。apache

--以上來自維基百科bootstrap

讓咱們用快遞員送快遞的例子來理解。vim

最開始送上門，你有時候不在家，又只有等次日，如今多了個菜鳥驛站，快遞小哥只須要往裏面放，你有快遞本身去驛站拿就好了。bash

快遞小哥也不用直接對接這麼多的客戶，有包裹往驛站丟就行，而後咱們去驛站拿，就算有時候咱們不空，也可讓包裹在驛站多放幾天，這也就是它的堆積能力。服務器

爲何咱們要用消息隊列

異步處理

假設你有一個系統調用的鏈路，系統A 調用系統B 耗時 50ms,系統B 調用系統C 又須要200ms ,系統C 調用系統 D ，須要作比較超時的操做，須要 2s,以下圖所示：網絡

如今上面最大的問題在於：一個用戶請求過來，整個鏈路的調用時間是 50ms + 200ms + 2000ms = 2250ms,也就是2秒多。

而事實上，調用鏈路中，系統A 調用系統 B，系統B 調用系統 C 總共加起來也才 250ms，可是系統C調用系統D 卻用了 2S。

正是加入系統C調用系統D 這個鏈路，致使系統響應時間從 250ms 增長到了 2250 ms，足足慢了 10倍。

若是說，咱們把系統D 從鏈路中抽離出去，讓 C 系統異步調用D，那麼在 B系統調用 C，C 處理完成本身的邏輯，發送一個異步的請求去調用D系統，不用阻賽等到 D系統響應了再返回。這是否是好不少了呢？

舉一個例子，就以咱們日常點外賣爲例：

咱們日常點完餐，付完款，系統而後平給帳戶扣款、建立訂單、通知商家準備菜品。

接着，是否是須要找個騎手給你送餐？那這個找騎手的過程，是須要一套複雜算法來實現調度的，比較耗時。

那麼咱們是否是就能夠把找騎手給你送餐的這個步驟從鏈路中抽離出去，作成異步化的，哪怕延遲個幾十秒，可是隻要在必定時間範圍內給你找到一個騎手去送餐就能夠了。

這樣是否是就可讓你下訂單點外賣的速度變得超快？支付成功以後，直接建立好訂單、帳戶扣款、通知商家立馬給你準備作菜就ok了，這個過程可能就幾百毫秒。而後後臺異步化的耗費可能幾十秒經過調度算法給你找到一個騎手去送餐，可是這個步驟不影響咱們快速下訂單。

因此上面的鏈路也是同理，若是業務流程支持異步化的話，是否是就能夠考慮把系統C對系統D的調用抽離出去作成異步化的，不要放在鏈路中同步依次調用。

整個過程以下：

消息解耦

假如你如今系統A，這個系統會產出一個核心數據，下游系統 B和 C 都須要這個數據。

那麼咱們日常作的就是直接調用系統 B 和系統 C，發送數據過去。

過程以下：

過幾天，其餘的業務系統D、E 也須要這個數據，而後成了這樣

若是後續還有系統要呢？你不應代碼改死。。。

這種狀況系統耦合很是的嚴重，若是你發送一個系統的調用失敗了怎麼整？

針對上面的問題，咱們可使用消息隊列來實現系統解藕。

系統A 把數據發送到消息隊列中，其餘的系統，誰須要，本身去消息隊列取就完了。

流量消峯&日誌收集

假設你有一個系統，平時正常的時候每秒可能就幾百個請求，系統部署在8核16G的機器的上，正常處理都是ok的，每秒幾百請求是能夠輕鬆抗住的。

可是以下圖所示，在高峯期一會兒來了每秒鐘幾千請求，瞬時出現了流量高峯，此時你的選擇是要搞10臺機器，抗住每秒幾千請求的瞬時高峯嗎？

那若是瞬時高峯天天就那麼半個小時，接着直接就下降爲了每秒就幾百請求，若是你線上部署了不少臺機器，那麼每臺機器就處理每秒幾十個請求就能夠了，這不是有點浪費機器資源嗎？

大部分時候，每秒幾百請求，一臺機器就足夠了，可是爲了抗那天天瞬時的高峯，硬是部署了10臺機器，天天就那半個小時有用，別的時候都是浪費資源的。

此時咱們就可使用消息隊列來幫忙了，進行消峯。全部機器前面部署一層MQ，平時每秒幾百請求你們均可以輕鬆接收消息。

一旦到了瞬時高峯期，一下涌入每秒幾千的請求，就能夠積壓在MQ裏面，而後那一臺機器慢慢的處理和消費。

等高峯期過了，再消費一段時間，MQ裏積壓的數據就消費完畢了。

以下圖：

以上內容來自於【石杉的架構筆記】，圖是自個畫的，算是本身的一些理解吧。

要了解他的更多內容，移步 juejin.im/user/5be058…

消息隊列有哪些

四種經常使用的消息隊列 ActiveMQ、RabbitMQ、RocketMQ、Kafka

juejin.im/post/5b41fe…

Kafka 是什麼

Kafka是一種高吞吐量的分佈式發佈訂閱消息系統，使用Scala編寫。

對於熟悉JMS（Java Message Service）規範的同窗來講，消息系統已經不是什麼新概念了(例如ActiveMQ，RabbitMQ等)。

Kafka擁有做爲一個消息系統應該具有的功能，可是確有着獨特的設計。能夠這樣來講，Kafka借鑑了JMS規範的思想，可是確並無徹底遵循JMS規範。

kafka是一個分佈式的，分區的消息(官方稱之爲commit log)服務。它提供一個消息系統應該具有的功能，可是確有着獨特的設計。首先，讓咱們來看一下基礎的消息(Message)相關術語：

Topic: Kafka按照Topic分類來維護消息
Producer：咱們將發佈(publish)消息到Topic的進程稱之爲生產者(producer)
Consumer：咱們將訂閱(subscribe)Topic而且處理Topic中消息的進程稱之爲消費者(consumer)
Broker： Kafka以集羣的方式運行，集羣中的每一臺服務器稱之爲一個代理(broker)。

所以，從一個較高的層面上來看，producers 經過網絡發送消息到Kafka集羣，而後consumers來進行消費，以下圖：

服務端(brokers)和客戶端(producer、consumer)之間通訊經過TCP協議來完成。咱們爲Kafka提供了一個Java客戶端，可是也可使用其餘語言編寫的客戶端。

Kafka的架構圖

Kafka的核心概念

Topic 和 Log

讓咱們首先深刻理解Kafka提出一個高層次的抽象概念 --Topic。

能夠理解Topic是一個類別的名稱，全部的message發送到Topic下面。對於每個Topic，kafka集羣按照以下方式維護一個分區(Partition,能夠就理解爲一個隊列Queue)日誌文件:

partition 是一個有序的 message 序列，這些 message 按順序添加到一個叫作 commit log 的文件中。每一個 partition 中的消息都有一個惟一的編號，稱之爲 offset，用來惟一標示某個分區中的message。

partition 支持消息位移讀取，消息位移有消費者自身管理，好比下圖：

由上圖能夠看出，不一樣消費者對同一分區的消息讀取互不干擾，消費者能夠經過設置消息位移（offset）來控制本身想要獲取的數據，好比：能夠從頭讀取，最新數據讀取，重讀讀取等功能。

Distribution

log 的 partitions 分佈在 kafka 集羣中不一樣的 broker 上，每一個 broker 能夠請求備份其餘broker 上 partition 上的數據。kafka 集羣支持配置一個 partition 備份的數量。針對每一個partition，都有一個 broker 起到「leader」的做用，0個多個其餘的 broker 做爲「follwers」的做用。

leader 處理全部的針對這個 partition 的讀寫請求，而 followers 被動複製 leader 的結果。若是這個 leader 失效了，其中的一個 follower 將會自動的變成新的 leader。

每一個 broker 都是本身所管理的 partition 的 leader，同時又是其餘 broker 所管理 partitions 的 followers，kafka 經過這種方式來達到負載均衡。

Producers

生產者將消息發送到topic中去，同時負責選擇將message發送到topic的哪個partition中。經過round-robin作簡單的負載均衡。也能夠根據消息中的某一個關鍵字來進行區分。一般第二種方式使用的更多。

Consumers

傳統的消息傳遞模式有2種：隊列( queuing)和（ publish-subscribe）。

在queuing模式中，多個consumer從服務器中讀取數據，消息只會到達一個consumer。

在 publish-subscribe 模型中，消息會被廣播給全部的consumer。

Kafka基於這2種模式提供了一種consumer的抽象概念：consumer group

每一個consumer都要標記本身屬於哪個consumer group。發佈到topic中的message中message會被傳遞到consumer group中的一個consumer 實例。consumer實例能夠運行在不一樣的進程上，也能夠在不一樣的物理機器上。

若是全部的consumer都位於同一個consumer group 下，這就相似於傳統的queue模式，並在衆多的consumer instance之間進行負載均衡。

若是全部的consumer都有着本身惟一的consumer group，這就相似於傳統的publish-subscribe模型。

更通常的狀況是，一般一個topic會有幾個consumer group，每一個consumer group都是一個邏輯上的訂閱者（ logical subscriber ）。每一個consumer group由多個consumer instance組成，從而達到可擴展和容災的功能。這並無什麼特殊的地方，僅僅是將publish-subscribe模型中的運行在單個進程上的consumers中的consumer替換成一個consumer group。以下圖所示：

說明：由2個broker組成的kafka集羣，總共有4個Parition(P0-P3)。這個集羣由2個Consumer Group， A有2個 consumer instances ，而B有四個.

消費順序

Kafka比傳統的消息系統有着更強的順序保證。在傳統的狀況下，服務器按照順序保留消息到隊列，若是有多個consumer來消費隊列中的消息，服務器會接受消息的順序向外提供消息。

可是，儘管服務器是按照順序提供消息，可是消息傳遞到每個consumer是異步的，這可能會致使先消費的 consumer獲取到消息時間可能比後消費的consumer獲取到消息的時間長，致使不能保證順序性。

這代表，當進行並行的消費的時候，消息在多個 consumer之間可能會失去順序性。

消息系統一般會採起一種「 exclusive consumer」的概念，來確保同一時間內只有一個consumer可以從隊列中進行消費，可是這實際上意味着在消息處理的過程當中是不支持並行的。

Kafka在這方面作的更好。經過Topic中並行度的概念，即partition，Kafka能夠同時提供順序性保證和多個consumer同時消費時的負載均衡。實現的原理是經過將一個topic中的partition分配給一個consumer group中的不一樣consumer instance。

經過這種方式，咱們能夠保證一個partition在同一個時刻只有一個consumer instance在消息，從而保證順序。

雖然一個topic中有多個partition，可是一個consumer group中同時也有多個consumer instance，經過合理的分配依然可以保證負載均衡。

須要注意的是，一個consumer group中的consumer instance的數量不能比一個Topic中的partition的數量多。若是多了，它將分配不上分區消息。

Kafka只在partition的範圍內保證消息消費的局部順序性，不能在同一個topic中的多個partition中保證總的消費順序性。一般來講，這已經能夠知足大部分應用的需求。

可是，若是的確有在整體上保證消費的順序的需求的話，那麼咱們能夠經過將topic的partition數量設置爲1，將consumer group中的consumer instance數量也設置爲1。

可是這樣作，Kafka 的吞吐量就會降低。

kafka的安裝與使用

安裝前的環境準備

因爲Kafka是用Scala語言開發的，運行在JVM上，所以在安裝Kafka以前須要先安裝JDK。

# yum install java-1.8.0-openjdk* -y
複製代碼

kafka依賴zookeeper，因此須要先安裝zookeeper

# wget http://mirror.bit.edu.cn/apache/zookeeper/stable/zookeeper-3.4.12.tar.gz
# tar -zxvf zookeeper-3.4.12.tar.gz
# cd zookeeper-3.4.12
# cp conf/zoo_sample.cfg conf/zoo.cfg 啓動zookeeper
# bin/zkServer.sh start
# bin/zkCli.sh
# ls / #查看zk的根目錄相關節點
複製代碼

第一步：下載安裝包

下載1.1.0 release版本，並解壓：

# wget https://archive.apache.org/dist/kafka/1.1.0/kafka_2.11-1.1.0.tgz
# tar -xzf kafka_2.11-1.1.0.tgz
# cd kafka_2.11-1.1.0
複製代碼

第二步：啓動服務

如今來啓動kafka服務：啓動腳本語法：kafka-server-start.sh [-daemon] server.properties

能夠看到，server.properties的配置路徑是一個強制的參數，-daemon表示之後臺進程運行，不然ssh客戶端退出後，就會中止服務。(注意，在啓動kafka時會使用linux主機名關聯的ip地址，因此須要把主機名和linux的ip映射配置到本地host裏，用vim /etc/hosts)

# bin/kafka-server-start.sh -daemon config/server.properties
複製代碼

咱們進入zookeeper目錄經過zookeeper客戶端查看下zookeeper的目錄樹

# bin/zkCli.sh
# ls / #查看zk的根目錄kafka相關節點
# ls /brokers/ids #查看kafka節點
複製代碼

第三步：建立主題

如今咱們來建立一個名字爲「test」的Topic，這個topic只有一個partition，而且備份因子也設置爲1：

# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
複製代碼

如今咱們能夠經過如下命令來查看kafka中目前存在的topic

# bin/kafka-topics.sh --list --zookeeper localhost:2181
複製代碼

除了咱們經過手工的方式建立Topic，咱們能夠配置broker，當producer發佈一個消息某個指定的Topic，可是這個Topic並不存在時，就自動建立。

第四步：發送消息

kafka自帶了一個producer命令客戶端，能夠從本地文件中讀取內容，或者咱們也能夠以命令行中直接輸入內容，並將這些內容以消息的形式發送到kafka集羣中。在默認狀況下，每個行會被當作成一個獨立的消息。

首先咱們要運行發佈消息的腳本，而後在命令中輸入要發送的消息的內容：

# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>this is a msg
>this is a another msg
複製代碼

第五步：消費消息

對於consumer，kafka一樣也攜帶了一個命令行客戶端，會將獲取到內容在命令中進行輸出：

# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning #老版本

# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testGroup --consumer-property client.id=consumer-1 --topic test #新版本
複製代碼

若是你是經過不一樣的終端窗口來運行以上的命令，你將會看到在producer終端輸入的內容，很快就會在consumer的終端窗口上顯示出來。

Kafka實操的一些命令

啓動zk 
bin/zkServer.sh start
啓動kafka
bin/kafka-server-start.sh config/server.properties &
中止kafka 若是無論用 就是用kill -9 
bin/kafka-server-stop.sh
1.建立主題
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2.列出主題
bin/kafka-topics.sh --list --zookeeper localhost:2181
3.生產消息
bin/kafka-console-producer.sh --broker-list localhost:9092 -topic test
4.消費消息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 -topic test --from-beginning
5.刪除主題
 1. 刪除 kafka 主題 
    bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic sceniccenter-base-ticket
 2. 在kafka 數據目錄刪除主題文件夾
 3. 刪除 zookeeper 上的 記錄
    1）登陸zookeeper客戶端：命令：./zkCli.sh 
        2）找到topic所在的目錄：ls /brokers/topics
        3）找到要刪除的topic，執行命令：rmr /brokers/topics/【topic name】便可，此時topic被完全刪除。    
    另外被標記爲 marked for deletion 的topic你能夠在zookeeper客戶端中經過命令得到： ls /admin/delete_topics/【topic name】
  總結  完全刪除topic：
     一、刪除kafka存儲目錄（server.properties文件log.dirs配置，默認爲"/tmp/kafka-logs"）相關topic目錄
     二、若是配置了delete.topic.enable=true直接經過命令刪除，若是命令刪除不掉，直接經過 zookeeper-client 刪除掉broker下的topic便可。
6.查看toplic 的分區等狀況 
bin/kafka-topics --describe --zookeeper hadoop1:2181 --topic wwcc1
Topic:wwcc1    PartitionCount:3    ReplicationFactor:3    Configs:
    Topic: wwcc1    Partition: 0    Leader: 127    Replicas: 127,128,129    Isr: 127,128,129
    Topic: wwcc1    Partition: 1    Leader: 128    Replicas: 128,129,127    Isr: 128,129,127
    Topic: wwcc1    Partition: 2    Leader: 129    Replicas: 129,127,128    Isr: 129,127,128

複製代碼