Kafka入门教程（转载）
原文地址：https://blog.csdn.net/dapeng1995/article/details/81536862
1.1?消息隊列（Message Queue)
Message Queue消息傳送系統(tǒng)提供傳送服務(wù)。消息傳送依賴于大量支持組件，這些組件負責處理連接服務(wù)、消息的路由和傳送、持久性、安全性以及日志記錄。消息服務(wù)器可以使用一個或多個代理實例。
JMS（Java Messaging Service）是Java平臺上有關(guān)面向消息中間件(MOM)的技術(shù)規(guī)范，它便于消息系統(tǒng)中的Java應(yīng)用程序進行消息交換,并且通過提供標準的產(chǎn)生、發(fā)送、接收消息的接口簡化企業(yè)應(yīng)用的開發(fā)，翻譯為Java消息服務(wù)。
?
1.2?MQ消息模型
KafkaMQ消息模型圖1-1
?
1.3?MQ消息隊列分類
消息隊列分類：點對點和發(fā)布/訂閱兩種：
1、點對點：
消息生產(chǎn)者生產(chǎn)消息發(fā)送到queue中，然后消息消費者從queue中取出并且消費消息。
消息被消費以后，queue中不再有存儲，所以消息消費者不可能消費到已經(jīng)被消費的消息。Queue支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費。
2、發(fā)布/訂閱：
消息生產(chǎn)者（發(fā)布）將消息發(fā)布到topic中，同時有多個消息消費者（訂閱）消費該消息。和點對點方式不同，發(fā)布到topic的消息會被所有訂閱者消費。
?
1.4?MQ消息隊列對比
1、RabbitMQ：支持的協(xié)議多，非常重量級消息隊列，對路由(Routing)，負載均衡(Loadbalance)或者數(shù)據(jù)持久化都有很好的支持。
?
2、ZeroMQ：號稱最快的消息隊列系統(tǒng)，尤其針對大吞吐量的需求場景，擅長的高級/復雜的隊列，但是技術(shù)也復雜，并且只提供非持久性的隊列。
?
3、ActiveMQ：Apache下的一個子項，類似ZeroMQ，能夠以代理人和點對點的技術(shù)實現(xiàn)隊列。
?
4、Redis：是一個key-Value的NOSql數(shù)據(jù)庫，但也支持MQ功能，數(shù)據(jù)量較小，性能優(yōu)于RabbitMQ，數(shù)據(jù)超過10K就慢的無法忍受。
?
1.5?Kafka簡介
Kafka是分布式發(fā)布-訂閱消息系統(tǒng),它最初由?LinkedIn?公司開發(fā)，使用?Scala語言編寫,之后成為?Apache?項目的一部分。在Kafka集群中，沒有“中心主節(jié)點”的概念，集群中所有的服務(wù)器都是對等的，因此，可以在不做任何配置的更改的情況下實現(xiàn)服務(wù)器的的添加與刪除，同樣的消息的生產(chǎn)者和消費者也能夠做到隨意重啟和機器的上下線。
?
Kafka消息系統(tǒng)生產(chǎn)者和消費者部署關(guān)系圖1-2
Kafka消息系統(tǒng)架構(gòu)圖1-3
?
1.6?Kafka術(shù)語介紹
1、消息生產(chǎn)者：即：Producer，是消息的產(chǎn)生的源頭，負責生成消息并發(fā)送到Kafka
服務(wù)器上。
?
2、消息消費者：即：Consumer，是消息的使用方，負責消費Kafka服務(wù)器上的消息。
?
3、主題：即：Topic，由用戶定義并配置在Kafka服務(wù)器，用于建立生產(chǎn)者和消息者之間的訂閱關(guān)系：生產(chǎn)者發(fā)送消息到指定的Topic下，消息者從這個Topic下消費消息。
?
4、消息分區(qū)：即：Partition，一個Topic下面會分為很多分區(qū)，例如：“kafka-test”這個Topic下可以分為6個分區(qū)，分別由兩臺服務(wù)器提供，那么通常可以配置為讓每臺服務(wù)器提供3個分區(qū)，假如服務(wù)器ID分別為0、1，則所有的分區(qū)為0-0、0-1、0-2和1-0、1-1、1-2。Topic物理上的分組，一個?topic可以分為多個?partition，每個?partition?是一個有序的隊列。partition中的每條消息都會被分配一個有序的?id（offset）。
?
5、Broker：即Kafka的服務(wù)器，用戶存儲消息，Kafa集群中的一臺或多臺服務(wù)器統(tǒng)稱為?broker。
?
6、消費者分組：Group，用于歸組同類消費者，在Kafka中，多個消費者可以共同消息一個Topic下的消息，每個消費者消費其中的部分消息，這些消費者就組成了一個分組，擁有同一個分組名稱，通常也被稱為消費者集群。
?
7、Offset：消息存儲在Kafka的Broker上，消費者拉取消息數(shù)據(jù)的過程中需要知道消息在文件中的偏移量，這個偏移量就是所謂的Offset。
?
1.7?Kafka中Broker
1、Broker：即Kafka的服務(wù)器，用戶存儲消息，Kafa集群中的一臺或多臺服務(wù)器統(tǒng)稱為?broker。
?
2、Message在Broker中通Log追加的方式進行持久化存儲。并進行分區(qū)（patitions)。
?
3、為了減少磁盤寫入的次數(shù),broker會將消息暫時buffer起來,當消息的個數(shù)(或尺寸)達到一定閥值時,再flush到磁盤,這樣減少了磁盤IO調(diào)用的次數(shù)。
?
4、Broker沒有副本機制，一旦broker宕機，該broker的消息將都不可用。Message消息是有多份的。
?
5、Broker不保存訂閱者的狀態(tài)，由訂閱者自己保存。
?
6、無狀態(tài)導致消息的刪除成為難題（可能刪除的消息正在被訂閱），kafka采用基于時間的SLA(服務(wù)水平保證)，消息保存一定時間（通常為7天）后會被刪除。
?
7、消息訂閱者可以rewind back到任意位置重新進行消費，當訂閱者故障時，可以選擇最小的offset(id)進行重新讀取消費消息。
?
1.8?Kafka的Message組成
1、Message消息：是通信的基本單位，每個?producer?可以向一個?topic（主題）發(fā)布一些消息。
?
2、Kafka中的Message是以topic為基本單位組織的，不同的topic之間是相互獨立的。每個topic又可以分成幾個不同的partition(每個topic有幾個partition是在創(chuàng)建topic時指定的)，每個partition存儲一部分Message。
?
3、partition中的每條Message包含了以下三個屬性：
offset ?????即：消息唯一標識:對應(yīng)類型：long
MessageSize?對應(yīng)類型：int32
data????? ??是message的具體內(nèi)容。
?
1.9?Kafka的Partitions分區(qū)
1、Kafka基于文件存儲.通過分區(qū)，可以將日志內(nèi)容分散到多個server上,來避免文件尺寸達到單機磁盤的上限，每個partiton都會被當前server(kafka實例)保存。
?
2、可以將一個topic切分多任意多個partitions，來消息保存/消費的效率。
?
3、越多的partitions意味著可以容納更多的consumer，有效提升并發(fā)消費的能力。
?
1.10?Kafka的Consumers
1、消息和數(shù)據(jù)消費者，訂閱?topics并處理其發(fā)布的消息的過程叫做?consumers。
?
2、在?kafka中,我們可以認為一個group是一個“訂閱者”，一個Topic中的每個partions，只會被一個“訂閱者”中的一個consumer消費，不過一個?consumer可以消費多個partitions中的消息（消費者數(shù)據(jù)小于Partions的數(shù)量時）。注意：kafka的設(shè)計原理決定，對于一個topic，同一個group中不能有多于partitions個數(shù)的consumer同時消費，否則將意味著某些consumer將無法得到消息。
?
3、一個partition中的消息只會被group中的一個consumer消息。每個group中consumer消息消費互相獨立。
?
1.11?Kafka的持久化
1、一個Topic可以認為是一類消息，每個topic將被分成多partition(區(qū)),每個partition在存儲層面是append log文件。任何發(fā)布到此partition的消息都會被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量），partition是以文件的形式存儲在文件系統(tǒng)中。
?
2、Logs文件根據(jù)broker中的配置要求,保留一定時間后刪除來釋放磁盤空間。
????????
Kafka消息分區(qū)Partition圖1-4
?
Partition：
?? Topic物理上的分組，一個?topic可以分為多個?partition，每個?partition?是一個有序的隊列。partition中的每條消息都會被分配一個有序的?id（offset）。
?
3、為數(shù)據(jù)文件建索引：稀疏存儲，每隔一定字節(jié)的數(shù)據(jù)建立一條索引。下圖為一個partition的索引示意圖：
???????
Kafka消息分區(qū)Partition索引圖1-5
1.12?Kafka的分布式實現(xiàn)：
?????
Kafka分布式關(guān)系圖1-6
????????
Kafka生產(chǎn)環(huán)境關(guān)系圖1-7
1.13?Kafka的通訊協(xié)議：
1、Kafka的Producer、Broker和Consumer之間采用的是一套自行設(shè)計基于TCP層的協(xié)議，根據(jù)業(yè)務(wù)需求定制，而非實現(xiàn)一套類似ProtocolBuffer的通用協(xié)議。
?
2、基本數(shù)據(jù)類型：（Kafka是基于Scala語言實現(xiàn)的，類型也是Scala中的數(shù)據(jù)類型）
?
定長數(shù)據(jù)類型：int8,int16,int32和int64，對應(yīng)到Java中就是byte, short, int和long。
?
變長數(shù)據(jù)類型：bytes和string。變長的數(shù)據(jù)類型由兩部分組成，分別是一個有符號整數(shù)N(表示內(nèi)容的長度)和N個字節(jié)的內(nèi)容。其中，N為-1表示內(nèi)容為null。bytes的長度由int32表示，string的長度由int16表示。
?
數(shù)組：數(shù)組由兩部分組成，分別是一個由int32類型的數(shù)字表示的數(shù)組長度N和N個元素。
?
3、Kafka通訊的基本單位是Request/Response。
?
4、基本結(jié)構(gòu)：
RequestOrResponse => MessageSize(RequestMessage | ResponseMessage)
| 名稱 | 類型 | 描術(shù) |
| MessageSize | int32 | 表示RequestMessage或者ResponseMessage的長度 ? |
| RequestMessage ResponseMessage | — | ? |
?
?
5、通訊過程：
客戶端打開與服務(wù)器端的Socket
往Socket寫入一個int32的數(shù)字(數(shù)字表示這次發(fā)送的Request有多少字節(jié))
服務(wù)器端先讀出一個int32的整數(shù)從而獲取這次Request的大小
然后讀取對應(yīng)字節(jié)數(shù)的數(shù)據(jù)從而得到Request的具體內(nèi)容
服務(wù)器端處理了請求后，也用同樣的方式來發(fā)送響應(yīng)。
?
6、RequestMessage結(jié)構(gòu)：
RequestMessage => ApiKey ApiVersionCorrelationId ClientId Request
| 名稱 | 類型 | 描術(shù) |
| ApiKey | int16 | 表示這次請求的API編號 |
| ApiVersion | int16 | 表示請求的API的版本，有了版本后就可以做到后向兼容 |
| CorrelationId | int32 | 由客戶端指定的一個數(shù)字唯一標示這次請求的id，服務(wù)器端在處理完請求后也會把同樣的CorrelationId寫到Response中，這樣客戶端就能把某個請求和響應(yīng)對應(yīng)起來了。 |
| ClientId | string | 客戶端指定的用來描述客戶端的字符串，會被用來記錄日志和監(jiān)控，它唯一標示一個客戶端。 |
| Request | — | Request的具體內(nèi)容。 |
?
7、ResponseMessage結(jié)構(gòu)：
ResponseMessage => CorrelationId Response
| 名稱 | 類型 | 描術(shù) |
| CorrelationId | int32 | 對應(yīng)Request的CorrelationId。 |
| Response | — | 對應(yīng)Request的Response，不同的Request的Response的字段是不一樣的。 |
??????
Kafka采用是經(jīng)典的Reactor(同步IO)模式，也就是1個Acceptor響應(yīng)客戶端的連接請求，N個Processor來讀取數(shù)據(jù)，這種模式可以構(gòu)建出高性能的服務(wù)器。
?
8、Message結(jié)構(gòu)：
Message:Producer生產(chǎn)的消息,鍵-值對
Message => Crc MagicByte Attributes KeyValue
| 名稱 | 類型 | 描術(shù) |
| CRC | int32 | 表示這條消息(不包括CRC字段本身)的校驗碼。 |
| MagicByte | int8 | 表示消息格式的版本，用來做后向兼容，目前值為0。 |
| Attributes | int8 | 表示這條消息的元數(shù)據(jù)，目前最低兩位用來表示壓縮格式。 |
| Key | bytes | 表示這條消息的Key，可以為null。 |
| Value | bytes | 表示這條消息的Value。Kafka支持消息嵌套，也就是把一條消息作為Value放到另外一條消息里面。 |
?
9、MessageSet結(jié)構(gòu)：
MessageSet:用來組合多條Message，它在每條Message的基礎(chǔ)上加上了Offset和MessageSize
MessageSet => [Offset MessageSize Message]
| 名稱 | 類型 | 描術(shù) |
| Offset | int64 | 它用來作為log中的序列號，Producer在生產(chǎn)消息的時候還不知道具體的值是什么，可以隨便填個數(shù)字進去。 |
| MessageSize | int32 | 表示這條Message的大小。 |
| Message | - | 表示這條Message的具體內(nèi)容，其格式見上一小節(jié)。 |
?
?
10、?????Request/Respone和Message/MessageSet的關(guān)系：
?
Request/Response是通訊層的結(jié)構(gòu)，和網(wǎng)絡(luò)的7層模型對比的話，它類似于TCP層。
?
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.14?數(shù)據(jù)傳輸?shù)氖聞?wù)定義：
1、at most once:最多一次,這個和JMS中”非持久化”消息類似.發(fā)送一次，無論成敗，將不會重發(fā)。
at most once:消費者fetch消息,然后保存offset，然后處理消息;當client保存offset之后，但是在消息處理過程中出現(xiàn)了異常，導致部分消息未能繼續(xù)處理.那么此后”未處理”的消息將不能被fetch到，這就是“atmost once”。
?
2、at least once:消息至少發(fā)送一次，如果消息未能接受成功，可能會重發(fā)，直到接收成功。
at least once:消費者fetch消息，然后處理消息，然后保存offset.如果消息處理成功之后，但是在保存offset階段zookeeper異常導致保存操作未能執(zhí)行成功，這就導致接下來再次fetch時可能獲得上次已經(jīng)處理過的消息，這就是“atleast once”，原因offset沒有及時的提交給zookeeper，zookeeper恢復正常還是之前offset狀態(tài)。
?
3、exactly once:消息只會發(fā)送一次。
exactly once: kafka中并沒有嚴格的去實現(xiàn)(基于2階段提交，事務(wù))，我們認為這種策略在kafka中是沒有必要的。
?
注：通常情況下“at-least-once”是我們首選。(相比at most once而言，重復接收數(shù)據(jù)總比丟失數(shù)據(jù)要好)。
?
1.15?學習Kafka推薦書籍：
1.?《Apache Kafka》
2.?《從Paxos到Zookeeper分布式一致性原理與實踐》
總結(jié)
以上是生活随笔為你收集整理的Kafka入门教程（转载）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： python实现WebsocketSer
- 下一篇： css布局，让侧边栏高度撑满，并且不会随