如何進(jìn)行Kafka學(xué)習(xí)-創(chuàng)新互聯(lián)

消息生產(chǎn)者生產(chǎn)消息發(fā)送到queue中，然后消息消費者從queue中取出并消費消息

注意：

  1.消息被消費以后，queue中不再有存儲，所以消息消費者不可能消費到已經(jīng)被消費的消息

  2.Queue支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費

消息生產(chǎn)者（發(fā)布）將消息發(fā)布到topic中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發(fā)布到topic的消息會被所有訂閱者消費

支持的協(xié)議多，非常重量級消息隊列，對路由（Routing），負(fù)載均衡（Load balance）或者數(shù)據(jù)持久化都有很好的支持

號稱最快的消息隊列系統(tǒng)，尤其針對大吞吐量的需求場景，擅長高級/復(fù)雜的隊列，但是技術(shù)也復(fù)雜，并且只提供非持久性隊列

Apache下的一個子項，類似ZeroMQ，能夠以代理人和點對點的技術(shù)實現(xiàn)隊列

一個key-value的NoSQL數(shù)據(jù)庫，但也支持MQ功能，數(shù)據(jù)量較小，性能優(yōu)于RabbitMQ，數(shù)據(jù)超過10K就慢的無法忍受

Kafka是分布式發(fā)布-訂閱消息系統(tǒng)。它最初由Linkedln公司開發(fā)，使用Scala語言編寫，之后成為Apache項目的一部分。Kafka是一個分布式的，可劃分的（分區(qū)處理），多訂閱者，冗余備份的持久性的日志服務(wù)。它主要用于處理活躍的流式數(shù)據(jù)

1.同時為發(fā)布和訂閱提供高吞吐量。據(jù)了解，Kafka每秒可以生產(chǎn)約25萬消息（50M），每秒處理55萬消息（110M）

2.可進(jìn)行持久化操作。將消息持久化到磁盤，因此可用于批量消費，例如ETL，以及實時應(yīng)用程序。通過將數(shù)據(jù)持久化到硬盤以及replication防止數(shù)據(jù)丟失

3.分布式系統(tǒng)，易于向外擴展。所有的producer、broker和consumer都會有多個，均為分布式。無需停機即可擴展機器

4.消息被處理的狀態(tài)是在consumer端維護(hù)，而不是由server端維護(hù)

5.支持online（上線）和offline（下線）的場景

重要說明：

1.在Kafka的體系中不存在單讀的Conmuser，它會存在一個Conmuser Group，Conmuser Group里面會有多個Conmuser

2.可以把Consumer Group看成一個虛擬的Consumer，它消費的是一個具體的Topic的數(shù)據(jù)，但具體執(zhí)行是由Consumer Group中的Consumer去執(zhí)行的，Consumer是一個邏輯上的概念，是不存在的，而存在的是Consumer Group當(dāng)中的Consumer， 一個Consumer Group對應(yīng)的是Topic，Consumer Group中的Consumer對應(yīng)的是Topic中的partition

3.一個消費者組里面的多個消費者對應(yīng)的是什么呢？

Topic組里面不同Partition的數(shù)據(jù)，一個Partition里面的數(shù)據(jù)交給一個Consumer來處理，另一個Partition里面的數(shù)據(jù)交給另一個Consumer來處理，當(dāng)然它們必須是同一個Consumer Group里面的Consumer，這就達(dá)到了并行的消費（每一個Consumer對應(yīng)的是一個Partition里面的數(shù)據(jù)）

4.Kafka為什么會有Partition的概念？

帶來的好處就是處理的速度更快，不同的Conmuser去消費不同Partition的消息，數(shù)據(jù)的消費就變成了并行的

特指消息的生產(chǎn)者

特指消息的消費者

消費者組，可以并行消費Topic中Partition的消息

緩存代理，Kafka集群中的一臺或多臺服務(wù)器統(tǒng)稱為broker

1.message在broker中通過log追加的方式進(jìn)行持久化存儲。并進(jìn)行分區(qū)（patitions）

2.為了減少磁盤寫入的次數(shù)，broker會將消息展示buffer起來，當(dāng)消息的個數(shù)（或尺寸）達(dá)到一定閥值時，再flush到磁盤，這樣減少了磁盤IO調(diào)用的次數(shù)

3.Broker沒有副本機制，一旦broker宕機，該broker的消息將不可用（但是消息是有副本的，可以把消息的副本同步到其它的broker中）

4.Broker不保存訂閱者的狀態(tài)，由訂閱者自己保存

5.無狀態(tài)導(dǎo)致消息的刪除成為難題（可能刪除的消息正在被訂閱）Kafka采用基于時間的SLA（服務(wù)水平保證），消息保存一定的時間（通常為7天）后會被刪除

6.消息訂閱者可以rewind back到任意位置重新進(jìn)行消費，當(dāng)訂閱者故障時，可以選擇最小的offset（id）進(jìn)行重新讀取消費消息

特指Kafka處理的消息源（feeds of messages）的不同分類

1.Topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）

2.Kafka的Partitions分區(qū)的目的

2.1 kafka基于文件存儲，通過分區(qū)，可以將日志內(nèi)容分線到多個server上，來避免文件尺寸達(dá)到單擊磁盤的上線，每個partition都會被當(dāng)前server（kafka實例）保存

2.2 可以將一個topic切分任意多個partitions來提高消息保存/消費的效率

2.3 越多的partitions意味著可以容納更多的consumer，有效提升并發(fā)消費的能力

1.消息，是通信的基本單位，每個producer可以向一個topic（主題）發(fā)布一些消息

2.Kafka中的Message是以topic為基本單位組織的，不同的topic之間是相互獨立的。每個topic有可以分成幾個不同的partition（每個topic有幾個partition是在創(chuàng)建topic時指定的）每個partition存儲一部分Message

3.partition中的每條Message包含了一下三個屬性

        屬性名稱           數(shù)據(jù)類型

        offset         long

        MessageSize          int32

        data          mssage的具體內(nèi)容

消息和數(shù)據(jù)生產(chǎn)者，向kafka的一個topic發(fā)布消息的過程叫做producers

1.producer將消息發(fā)布到指定的topic中，同時producer也能決定將此消息歸屬于哪個partition。比如基于“round-robin”方式或者通過其他的一些算法等

2.異步發(fā)送：批量發(fā)送可以很有效的提高發(fā)送效率。kafka producer的異步發(fā)送模式允許進(jìn)行批量發(fā)送，先將消息緩存在內(nèi)存中，然后一次請求批量發(fā)送出去

1.消息和數(shù)據(jù)消費者，訂閱topics并處理其發(fā)布的消息的過程叫做consumers

2.在Kafka中，我們可以認(rèn)為一個group是一個“訂閱者”，一個Topic中的每個partition只會被一個“訂閱者”中的一個consumer消費，不過一個consumer可以消費多個partition中的消息（消費者數(shù)據(jù)小于Partition的數(shù)量時）

3.注意：kafka的設(shè)計原理決定，對于一個topic，同一個group中不能有多于partitions個數(shù)的consumer同時消費，否則將意味著某些consumer將無法得到消息

1.發(fā)現(xiàn)線性的訪問磁盤，很多時候比隨機的內(nèi)存訪問快的多

2.傳統(tǒng)的使用內(nèi)存作為磁盤的緩存

3.Kafka直接將數(shù)據(jù)寫入到日志文件

1.寫操作：通過將數(shù)據(jù)追加到文件中實現(xiàn)

2.讀操作：讀的時候從文件讀就好了

1.讀操作不會阻塞寫稻作和其它操作，數(shù)據(jù)大小不對性能產(chǎn)生影響

2.沒有容量限制（相對于內(nèi)存來說）的硬盤空間建立消息系統(tǒng)

3.線性訪問磁盤，速度快，可以保存任意一段時間

1.一個Tipic可以認(rèn)為是一個類消息，每個topic將被分成多個partition，每個partition在存儲層面是append log文件。任何發(fā)布到此partition的消息都會被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量），partition是以文件的形式存儲在文件系統(tǒng)中

2.Logs文件根據(jù)broker中的配置要求，保留一定時間后刪除來釋放磁盤空間（默認(rèn)是7天）

說明：Partition是Topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。Partition中每條消息都會被分配一個有序的id（offset）

為數(shù)據(jù)文件建立索引：稀疏存儲，每隔一定字節(jié)的數(shù)據(jù)建立一條索引。下圖是一個partition的索引示意圖

注意：

1. 現(xiàn)在對1、3、6、8 建立了索引，如果要查找7，則會先查找到8然后，再找到8后的一個索引6，然后兩個索引之間做二分法，找到7的位置

2. 日志文件也會進(jìn)行segement（分割），分而治之

注意：

1.當(dāng)生產(chǎn)者將消息發(fā)送到Kafka后，就會去立刻通知ZooKeeper，zookeeper中會watch到相關(guān)的動作，當(dāng)watch到相關(guān)的數(shù)據(jù)變化后，會通知消費者去消費消息

2.消費者是主動去Pull（拉）kafka中的消息，這樣可以降低Broker的壓力，因為Broker中的消息是無狀態(tài)的，Broker也不知道哪個消息是可以消費的

3.當(dāng)消費者消費了一條消息后，也必須要去通知ZooKeeper。zookeeper會記錄下消費的數(shù)據(jù)，這樣但系統(tǒng)出現(xiàn)問題后就可以還原，可以知道哪些消息已經(jīng)被消費了

說明：

1.Name Server集群指的是Zookeeper集群

1.Kafka的通訊協(xié)議主要說的是，consumer去拉數(shù)據(jù)使用的通訊協(xié)議

2.Kafka的Producer、Broker和Consumer采用的是一套自行設(shè)計基于TCP層的協(xié)議，根據(jù)業(yè)務(wù)需求定制，而非實現(xiàn)一套類似于Protocol Buffer的通訊協(xié)議

3.基本數(shù)據(jù)類型

3.1定長數(shù)據(jù)類型：int8，int16，int32和int64，對應(yīng)到Java中就是byte，short，int和long

3.2變長數(shù)據(jù)類型：bytes和string。變長的數(shù)據(jù)類型由兩部分組成，分別是一個有符號整數(shù)N（標(biāo)識內(nèi)容的長度）和N個字節(jié)的內(nèi)容。其中N為-1標(biāo)識內(nèi)容為null。Bytes的長度由int32標(biāo)識，string的長度由int16表示

3.3數(shù)組：數(shù)組由兩個部分組成，分別是一個有int32類型的數(shù)字標(biāo)識的數(shù)組長度N和N個元素

1.Kafka通訊的基本單位是Request/Response

2.基本結(jié)構(gòu)：

  RequestOrResponse ---> MessageSize(RequestMessage | ResponseMessage)

3.通訊過程：

3.1客戶端打開與服務(wù)端的Socket

3.2往Socket寫入一個int32的數(shù)字（數(shù)字標(biāo)識這次發(fā)送的Request有多少字節(jié)）

3.3服務(wù)器端先讀出一個int32的整數(shù)從而獲取這次Request的大小

3.4然后讀取對應(yīng)字節(jié)數(shù)的數(shù)據(jù)從而得到Request的具體內(nèi)容

3.5服務(wù)器端處理了請求之后也用同樣的發(fā)送發(fā)誓來發(fā)送響應(yīng)

4.RequestMessage結(jié)構(gòu)

4.1RequestMessage ---> ApiKey ApiVersion CorrelationId ClientId Request

5.ResponseMessage

5.1ResponseMessage --->  CorrelationId Response

Kafka采用是經(jīng)典的Reactor（同步IO）模式，也就是1個Acceptor響應(yīng)客戶端的連接請求，N個Processor來讀取數(shù)據(jù)，這種模式可以構(gòu)建出高性能的服務(wù)器

6.Message：Producer生產(chǎn)的消息，鍵-值對

6.1Message --- > Crc MagicByte Attributes Key Value

說明：

CRC是一種消息檢驗方式，在Consumer拿到數(shù)據(jù)以后，CRC會獲取MessageSize和MessageData的大小做比較，如果不一致則，那么這個操作的數(shù)據(jù)Consumer就不接收了，如果一直則才做處理。防止消息在傳輸過程中損壞，丟失的一種校驗方式

7.MessageSet：用來組合多條Message，它在每條Message的基礎(chǔ)上加上offset和MessageSize

7.1MessageSet --> [offset MessageSize Message]

8.Request/Response和Message/messageSet的關(guān)系

8.1 Request/Response是通訊層的結(jié)構(gòu)，和網(wǎng)絡(luò)的7層模型對比的話，它類似于TCP

8.2 Message/MessageSet定義的是業(yè)務(wù)層的結(jié)構(gòu)，類似于網(wǎng)絡(luò)7層模型中的HTTP層。Message/MessageSet只是Request/Response的payload中的一種數(shù)據(jù)結(jié)構(gòu)

備注：Kafka的通訊協(xié)議中不包含Schema，格式也比較簡單，這樣設(shè)計的好處是協(xié)議自身的Overhead小，再加上把多條Message放到一期做壓縮，提高壓縮比率，從而在網(wǎng)絡(luò)上傳輸?shù)臄?shù)據(jù)量會少一些

1.at most once：最多一次，這個和JMS中“非持久化”消息類似，發(fā)送一次，無論成敗，將不會重發(fā)

消費者fetch（得到）消息，然后保存offset，然后處理消息；  當(dāng)client保存offset之后，但是在消息處理過程中出現(xiàn)了異常，導(dǎo)致部分消息未能繼續(xù)處理，那么伺候“未處理”的消息將不能被fetch到，這就是“at most once”

2.at least once：消息至少發(fā)送一次，如果消息未能接收成功，可能會重發(fā)，知道接收成功

消費者fetch消息，然后處理消息，然后保存offset，如果消息處理成功之后，但是在保存offset階段zookeeper異常導(dǎo)致保存操作未能執(zhí)行成功，這就導(dǎo)致接下來再次fetch時可能獲得上次已經(jīng)處理過的消息，這就是“at least once”，原因offset沒有即使的提交給zookeeper，zookeeper恢復(fù)正常還是之前offset狀態(tài)。

注：通常情況下“at least once”是我們的選（相比at most once而言，重復(fù)接收數(shù)據(jù)總比丟失數(shù)據(jù)要好）

3.exactly once：消息只會發(fā)送一次

Kafka中并沒有嚴(yán)格的去實現(xiàn)（基于2階段提交，事務(wù)），我們認(rèn)為這種策略在kafka中是沒有必要的

1.下載并上傳kafka到服務(wù)器

2.解壓縮并移動到/usr/local目錄下

3.啟動服務(wù)

3.1啟動zookeeper服務(wù) # ./zookeeper-server-start.sh ../config/zookeeper.properties > /dev/null 2>&1 &

3.2啟動kafka服務(wù) # ./kafka-server-start.sh ../config/server.properties > /dev/null 2>&1 &

3.3創(chuàng)建topic：

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

3.4查看主題

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

3.5查看主題詳情

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

3.6刪除主題

./kafka-run-class.sh kafka.admin.TopicCommand --delete --topic test --zookeeper 192.168.31.220:2181

./kafka-console-producer.sh --broker-list localhost:9092 --topic test1

./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test1 --from-beginning

生產(chǎn)者參數(shù)查看：./kafka-console-producer.sh

消費者參數(shù)查看：./kafka-console-consumer.sh