消息生產者生產消息發送到queue中，然后消息消費者從queue中取出并消費消息

注意：

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

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

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

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

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

Apache下的一個子項，類似ZeroMQ，能夠以代理人和點對點的技術實現隊列

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

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

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

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

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

4.消息被處理的狀態是在consumer端維護，而不是由server端維護

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

重要說明：

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

2.可以把Consumer Group看成一個虛擬的Consumer，它消費的是一個具體的Topic的數據，但具體執行是由Consumer Group中的Consumer去執行的，Consumer是一個邏輯上的概念，是不存在的，而存在的是Consumer Group當中的Consumer， 一個Consumer Group對應的是Topic，Consumer Group中的Consumer對應的是Topic中的partition

3.一個消費者組里面的多個消費者對應的是什么呢？

Topic組里面不同Partition的數據，一個Partition里面的數據交給一個Consumer來處理，另一個Partition里面的數據交給另一個Consumer來處理，當然它們必須是同一個Consumer Group里面的Consumer，這就達到了并行的消費（每一個Consumer對應的是一個Partition里面的數據）

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

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

特指消息的生產者

特指消息的消費者

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

緩存代理，Kafka集群中的一臺或多臺服務器統稱為broker

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

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

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

4.Broker不保存訂閱者的狀態，由訂閱者自己保存

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

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

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

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

2.Kafka的Partitions分區的目的

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

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

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

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

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

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

        屬性名稱           數據類型

        offset                long

        MessageSize          int32

        data                  mssage的具體內容

消息和數據生產者，向kafka的一個topic發布消息的過程叫做producers

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

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

1.消息和數據消費者，訂閱topics并處理其發布的消息的過程叫做consumers

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

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

1.發現線性的訪問磁盤，很多時候比隨機的內存訪問快的多

2.傳統的使用內存作為磁盤的緩存

3.Kafka直接將數據寫入到日志文件

1.寫操作：通過將數據追加到文件中實現

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

1.讀操作不會阻塞寫稻作和其它操作，數據大小不對性能產生影響

2.沒有容量限制（相對于內存來說）的硬盤空間建立消息系統

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

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

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

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

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

注意：

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

2. 日志文件也會進行segement（分割），分而治之

注意：

1.當生產者將消息發送到Kafka后，就會去立刻通知ZooKeeper，zookeeper中會watch到相關的動作，當watch到相關的數據變化后，會通知消費者去消費消息

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

3.當消費者消費了一條消息后，也必須要去通知ZooKeeper。zookeeper會記錄下消費的數據，這樣但系統出現問題后就可以還原，可以知道哪些消息已經被消費了

說明：

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

1.Kafka的通訊協議主要說的是，consumer去拉數據使用的通訊協議

2.Kafka的Producer、Broker和Consumer采用的是一套自行設計基于TCP層的協議，根據業務需求定制，而非實現一套類似于Protocol Buffer的通訊協議

3.基本數據類型

3.1定長數據類型：int8，int16，int32和int64，對應到Java中就是byte，short，int和long

3.2變長數據類型：bytes和string。變長的數據類型由兩部分組成，分別是一個有符號整數N（標識內容的長度）和N個字節的內容。其中N為-1標識內容為null。Bytes的長度由int32標識，string的長度由int16表示

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

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

2.基本結構：

  RequestOrResponse ---> MessageSize(RequestMessage | ResponseMessage)

3.通訊過程：

3.1客戶端打開與服務端的Socket

3.2往Socket寫入一個int32的數字（數字標識這次發送的Request有多少字節）

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

3.4然后讀取對應字節數的數據從而得到Request的具體內容

3.5服務器端處理了請求之后也用同樣的發送發誓來發送響應

4.RequestMessage結構

4.1RequestMessage ---> ApiKey ApiVersion CorrelationId ClientId Request

5.ResponseMessage

5.1ResponseMessage --->  CorrelationId Response

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

6.Message：Producer生產的消息，鍵-值對

6.1Message --- > Crc MagicByte Attributes Key Value

說明：

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

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

7.1MessageSet --> [offset MessageSize Message]

8.Request/Response和Message/messageSet的關系

8.1 Request/Response是通訊層的結構，和網絡的7層模型對比的話，它類似于TCP

8.2 Message/MessageSet定義的是業務層的結構，類似于網絡7層模型中的HTTP層。Message/MessageSet只是Request/Response的payload中的一種數據結構

備注：Kafka的通訊協議中不包含Schema，格式也比較簡單，這樣設計的好處是協議自身的Overhead小，再加上把多條Message放到一期做壓縮，提高壓縮比率，從而在網絡上傳輸的數據量會少一些

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

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

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

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

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

3.exactly once：消息只會發送一次

Kafka中并沒有嚴格的去實現（基于2階段提交，事務），我們認為這種策略在kafka中是沒有必要的

1.下載并上傳kafka到服務器

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

3.啟動服務

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

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

3.3創建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

生產者參數查看：./kafka-console-producer.sh

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