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

注意：

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

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

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

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

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

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

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

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

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

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

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

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

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

重要說明：

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

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

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

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

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

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

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

特指消息的消費(fèi)者

消費(fèi)者組，可以并行消費(fèi)Topic中Partition的消息

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        offset         long

        MessageSize          int32

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

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

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

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

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

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

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

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

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

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

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

2.讀操作：讀的時(shí)候從文件讀就好了

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

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

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

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

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

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

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

注意：

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

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

注意：

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

2.消費(fèi)者是主動(dòng)去Pull（拉）kafka中的消息，這樣可以降低Broker的壓力，因?yàn)锽roker中的消息是無狀態(tài)的，Broker也不知道哪個(gè)消息是可以消費(fèi)的

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

說明：

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

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

2.Kafka的Producer、Broker和Consumer采用的是一套自行設(shè)計(jì)基于TCP層的協(xié)議，根據(jù)業(yè)務(wù)需求定制，而非實(shí)現(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ù)類型由兩部分組成，分別是一個(gè)有符號整數(shù)N（標(biāo)識內(nèi)容的長度）和N個(gè)字節(jié)的內(nèi)容。其中N為-1標(biāo)識內(nèi)容為null。Bytes的長度由int32標(biāo)識，string的長度由int16表示

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

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

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

  RequestOrResponse ---> MessageSize(RequestMessage | ResponseMessage)

3.通訊過程：

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

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

3.3服務(wù)器端先讀出一個(gè)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個(gè)Acceptor響應(yīng)客戶端的連接請求，N個(gè)Processor來讀取數(shù)據(jù)，這種模式可以構(gòu)建出高性能的服務(wù)器

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

6.1Message --- > Crc MagicByte Attributes Key Value

說明：

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

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è)計(jì)的好處是協(xié)議自身的Overhead小，再加上把多條Message放到一期做壓縮，提高壓縮比率，從而在網(wǎng)絡(luò)上傳輸?shù)臄?shù)據(jù)量會少一些

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

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

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

消費(fèi)者fetch消息，然后處理消息，然后保存offset，如果消息處理成功之后，但是在保存offset階段zookeeper異常導(dǎo)致保存操作未能執(zhí)行成功，這就導(dǎo)致接下來再次fetch時(shí)可能獲得上次已經(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)格的去實(shí)現(xiàn)（基于2階段提交，事務(wù)），我們認(rèn)為這種策略在kafka中是沒有必要的

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

2.解壓縮并移動(dòng)到/usr/local目錄下

3.啟動(dòng)服務(wù)

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

3.2啟動(dòng)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

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