【云小課】EI第37課 MRS基礎原理之Kafka組件介紹

Kafka是一個分布式的、分區的、多副本的消息發布-訂閱系統，它提供了類似于JMS的特性，但在設計上完全不同，它具有消息持久化、高吞吐、分布式、多客戶端支持、實時等特性，適用于離線和在線的消息消費，如常規的消息收集、網站活性跟蹤、聚合統計系統運營數據（監控數據）、日志收集等大量數據的互聯網服務的數據收集場景。

Kafka主要特點如下：

可靠性：提供At-Least Once，At-Most Once，Exactly Once消息可靠傳遞。消息被處理的狀態是在Consumer端維護，需要結合應用層實現Exactly Once。

高吞吐：同時為發布和訂閱提供高吞吐量。

持久化：將消息持久化到磁盤，因此可用于批量消費，以及實時應用程序。通過將數據持久化到硬盤以及replication防止數據丟失。

分布式：分布式系統，易于向外擴展。所有的Producer、Broker和Consumer都支持部署多個形成分布式的集群。無需停機即可擴展系統。

Kafka結構

生產者（Producer）將消息發布到Kafka主題（Topic）上，消費者（Consumer）訂閱這些主題并消費這些消息。在Kafka集群上一個服務器稱為一個Broker。對于每一個主題，Kafka集群保留一個用于縮放、并行化和容錯性的分區（Partition）。每個分區是一個有序、不可變的消息序列，并不斷追加到提交日志文件。分區的消息每個也被賦值一個稱為偏移順序（Offset）的序列化編號。

名稱

說明

Broker

在Kafka集群上一個服務器稱為一個Broker。

Topic/主題

一個Topic就是一個類別或者一個可訂閱的條目名稱，也即一類消息。一個主題可以有多個分區，這些分區可以作為并行的一個單元。

Partition/分區

是一個有序的、不可變的消息序列，這個序列可以被連續地追加—個提交日志。在分區內的每條消息都有一個有序的ID號，這個ID號被稱為偏移（Offset），這個偏移量可以唯一確定每條消息在分區內的位置。

Producer/生產者

向Kafka的主題發布消息。

Consumer/消費者

向Topic訂閱，并且接收發布到這些Topic的消息。

消費者使用一個消費者組名稱來標記自己，主題的每個消息被傳遞給每個訂閱消費者組中的一個消費者。如果所有的消費者實例都屬于同樣的消費組，它們就以傳統隊列負載均衡方式工作。如上圖中，Consumer1與Consumer2之間為負載均衡方式；Consumer3、Consumer4、Consumer5與Consumer6之間為負載均衡方式。如果消費者實例都屬于不同的消費組，則消息會被廣播給所有消費者。如上圖中，Topic1中的消息，同時會廣播到Consumer Group1與Consumer Group2中。

Kafka原理

消息可靠性

Kafka Broker收到消息后，會持久化到磁盤，同時，Topic的每個Partition有自己的Replica（備份），每個Replica分布在不同的Broker節點上，以保證當某一節點失效時，可以自動故障轉移到可用消息節點。

高吞吐量

Kafka通過以下方式提供系統高吞吐量：

數據磁盤持久化：消息不在內存中cache，直接寫入到磁盤，充分利用磁盤的順序讀寫性能。

Zero-copy：減少IO操作步驟。

數據批量發送：提高網絡利用率。

Topic劃分為多個Partition，提高并發度，可以由多個Producer、Consumer數目之間的關系并發來讀、寫消息。Producer根據用戶指定的算法，將消息發送到指定的Partition。

消息訂閱-通知機制

消費者對感興趣的主題進行訂閱，并采取pull的方式消費數據，使得消費者可以根據其消費能力自主地控制消息拉取速度，同時，可以根據自身情況自主選擇消費模式，例如批量、重復消費，從尾端開始消費等；另外，需要消費者自己負責維護其自身消息的消費記錄。

可擴展性

當在Kafka集群中可通過增加Broker節點以提供更大容量時。新增的Broker會向ZooKeeper注冊，而Producer及Consumer會及時從ZooKeeper感知到這些變化，并及時作出調整。

使用Kafka客戶端

MRS集群的創建您可參考創建集群。

Kafka客戶端的安裝您可以參考安裝客戶端。

以客戶端安裝用戶，登錄安裝客戶端的節點。執行以下命令，切換到客戶端安裝目錄。

cd /opt/Bigdata/client source bigdata_env

若集群開啟了Kerberos認證，需提前準備具有Kafka操作權限的用戶并進行認證。

kinit 組件業務用戶

創建一個Topic：

cd Kafka/kafka/bin/ sh kafka-topics.sh --create --topic 主題名稱 --partitions 主題占用的分區數 --replication-factor 主題的備份個數 --zookeeper ZooKeeper角色實例所在節點IP地址:clientPort/kafka 例如執行： sh kafka-topics.sh --create --topic Test --partitions 2 --replication-factor 2 --zookeeper 192.168.67.136:24002/kafka

執行以下命令，查詢集群中的Topic信息：

sh kafka-topics.sh --list --zookeeper ZooKeeper角色實例所在節點IP地址:ZooKeeper端口/kafka 例如執行： sh kafka-topics.sh --list --zookeeper 192.168.67.136:24002/kafka ... Test __KafkaMetricReport __consumer_offsets __default_metrics ...

登錄集群的FusionInsight Manager，單擊“集群 > 服務 > Kafka > KafkaTopic監控”可從管理界面中快速查看Topic狀態。

繼續執行以下命令，向Test Topic中寫入數據。

cat ../NOTICE | kafka-console-producer.sh --broker-list Broker實例IP地址:Broker端口 --topic Topic名稱 --producer.config ../config/producer.properties 例如執行： cat ../NOTICE | kafka-console-producer.sh --broker-list 192.168.42.14:21007 --topic Test --producer.config ../config/producer.properties

將Test Topic中的內容讀取出來。

kafka-console-consumer.sh --consumer.config ../config/consumer.properties --bootstrap-server Broker實例IP地址:Broker端口 --topic Topic名稱 --from-beginning 例如執行： kafka-console-consumer.sh --consumer.config ../config/consumer.properties --bootstrap-server 192.168.42.14:21007 --topic Test --from-beginning

Apache Kafka Copyright 2019 The Apache Software Foundation. This product includes software developed at The Apache Software Foundation (https://www.apache.org/). This distribution has a binary dependency on jersey, which is available under the CDDL License. The source code of jersey can be found at https://github.com/jersey/jersey/. ...

退出顯示窗口，執行以下命令刪除創建的Topic：

sh kafka-topics.sh --delete --topic Test --zookeeper 192.168.67.136:24002/kafka

關于Kafka應用開發及相關樣例代碼介紹，請參考《Kafka開發指南》。

好了，本期云小課就介紹到這里，快去體驗MapReduce(MRS)更多功能吧！猛戳這里

Kafka 云小課 大數據

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。