Kafka集羣安裝部署、Kafka生產者、Kafka消費者

時間 2019-11-08

標籤 kafka 集羣安裝部署生產者消費者欄目 Kafka 简体版

原文原文鏈接

Storm上游數據源之Kakfa

目標：html

理解Storm消費的數據來源、理解JMS規範、理解Kafka核心組件、掌握Kakfa生產者API、掌握Kafka消費者API。對流式計算的生態環境有深刻的瞭解，具有流式計算項目架構的能力。mysql

大綱：linux

一、 kafka是什麼？sql

二、 JMS規範是什麼？shell

三、爲何須要消息隊列？apache

四、 Kafka核心組件後端

五、 Kafka安裝部署數組

六、 Kafka生產者Java API緩存

七、 Kafka消費者Java API服務器

內容

一、Kafka是什麼

在流式計算中，Kafka通常用來緩存數據，Storm經過消費Kafka的數據進行計算。

KAFKA + STORM +REDIS

l Apache Kafka是一個開源消息系統，由Scala寫成。是由Apache軟件基金會開發的一個開源消息系統項目。

l Kafka最初是由LinkedIn開發，並於2011年初開源。2012年10月從Apache Incubator畢業。該項目的目標是爲處理實時數據提供一個統1、高通量、低等待的平臺。

l Kafka是一個分佈式消息隊列：生產者、消費者的功能。它提供了相似於JMS的特性，可是在設計實現上徹底不一樣，此外它並非JMS規範的實現。

l Kafka對消息保存時根據Topic進行歸類，發送消息者稱爲Producer,消息接受者稱爲Consumer,此外kafka集羣有多個kafka實例組成，每一個實例(server)成爲broker。

l 不管是kafka集羣，仍是producer和consumer都依賴於zookeeper集羣保存一些meta信息，來保證系統可用性

二、JMS是什麼

2.一、JMS的基礎

JMS是什麼：JMS是Java提供的一套技術規範

JMS幹什麼用：用來異構系統集成通訊，緩解系統瓶頸，提升系統的伸縮性加強系統用戶體驗，使得系統模塊化和組件化變得可行並更加靈活

經過什麼方式：生產消費者模式（生產者、服務器、消費者）

jdk，kafka，activemq……

2.二、JMS消息傳輸模型

l 點對點模式（一對一，消費者主動拉取數據，消息收到後消息清除）

點對點模型一般是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特色是發送到隊列的消息被一個且只有一個接收者接收處理，即便有多個消息監聽者也是如此。

l 發佈/訂閱模式（一對多，數據生產後，推送給全部訂閱者）

發佈訂閱模型則是一個基於推送的消息傳送模型。發佈訂閱模型能夠有多種不一樣的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的全部消息，即時當前訂閱者不可用，處於離線狀態。

queue.put（object）數據生產

queue.take(object) 數據消費

2.三、JMS核心組件

l Destination：消息發送的目的地，也就是前面說的Queue和Topic。

l Message [m1] ：從字面上就能夠看出是被髮送的消息。

l Producer：消息的生產者，要發送一個消息，必須經過這個生產者來發送。

l MessageConsumer：與生產者相對應，這是消息的消費者或接收者，經過它來接收一個消息。

經過與ConnectionFactory能夠得到一個connection

經過connection能夠得到一個session會話。

2.四、常見的類JMS消息服務器

2.4.一、JMS消息服務器 ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個徹底支持JMS1.1和J2EE 1.4規範的。

主要特色：

l 多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP

l 徹底支持JMS1.1和J2EE 1.4規範 (持久化,XA消息,事務)

l 對Spring的支持,ActiveMQ能夠很容易內嵌到使用Spring的系統裏面去,並且也支持Spring2.0的特性

l 經過了常見J2EE服務器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中經過JCA 1.5 resource adaptors的配置,可讓ActiveMQ能夠自動的部署到任何兼容J2EE 1.4 商業服務器上

l 支持多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

l 支持經過JDBC和journal提供高速的消息持久化

l 從設計上保證了高性能的集羣,客戶端-服務器,點對點

l 支持Ajax

l 支持與Axis的整合

l 能夠很容易得調用內嵌JMS provider,進行測試

2.4.二、分佈式消息中間件 Metamorphosis

Metamorphosis (MetaQ) 是一個高性能、高可用、可擴展的分佈式消息中間件，相似於LinkedIn的Kafka，具備消息存儲順序寫、吞吐量大和支持本地和XA事務等特性，適用於大吞吐量、順序消息、廣播和日誌數據傳輸等場景，在淘寶和支付寶有着普遍的應用，現已開源。

主要特色：

l 生產者、服務器和消費者均可分佈

l 消息存儲順序寫

l 性能極高,吞吐量大

l 支持消息順序

l 支持本地和XA事務

l 客戶端pull，隨機讀,利用sendfile系統調用，zero-copy ,批量拉數據

l 支持消費端事務

l 支持消息廣播模式

l 支持異步發送消息

l 支持http協議

l 支持消息重試和recover

l 數據遷移、擴容對用戶透明

l 消費狀態保存在客戶端

l 支持同步和異步複製兩種HA

l 支持group commit

2.4.三、分佈式消息中間件 RocketMQ

RocketMQ 是一款分佈式、隊列模型的消息中間件，具備如下特色：

l 可以保證嚴格的消息順序

l 提供豐富的消息拉取模式

l 高效的訂閱者水平擴展能力

l 實時的消息訂閱機制

l 億級消息堆積能力

l Metaq3.0 版本更名，產品名稱改成RocketMQ

2.4.四、其餘MQ

l .NET消息中間件 DotNetMQ

l 基於HBase的消息隊列 HQueue

l Go 的 MQ 框架 KiteQ

l AMQP消息服務器 RabbitMQ

l MemcacheQ 是一個基於 MemcacheDB 的消息隊列服務器。

三、爲何須要消息隊列（重要）

消息系統的核心做用就是三點：解耦，異步和並行

以用戶註冊的案列來講明消息系統的做用

3.一、用戶註冊的通常流程

問題：隨着後端流程愈來愈多，每步流程都須要額外的耗費不少時間，從而會致使用戶更長的等待延遲。

3.二、用戶註冊的並行執行

問題：系統並行的發起了4個請求，4個請求中，若是某一個環節執行1分鐘，其餘環節再快，用戶也須要等待1分鐘。若是其中一個環節異常以後，整個服務掛掉了。

3.三、用戶註冊的最終一致

一、 保證主流程的正常執行、執行成功以後，發送MQ消息出去。

二、須要這個destination的其餘系統經過消費數據再執行，最終一致。

四、Kafka核心組件

l Topic ：消息根據Topic進行歸類

l Producer：發送消息者

l Consumer：消息接受者

l broker：每一個kafka實例(server)

l Zookeeper：依賴集羣保存meta信息。

五、Kafka集羣部署

5.1集羣部署的基本流程

下載安裝包、解壓安裝包、修改配置文件、分發安裝包、啓動集羣

5.2集羣部署的基礎環境準備

安裝前的準備工做（zk集羣已經部署完畢）

l 關閉防火牆

chkconfig iptables off && setenforce 0

l 建立用戶

groupadd realtime &&　useradd realtime　&& usermod -a -G realtime realtime

l 建立工做目錄並賦權

mkdir /export

mkdir /export/servers

chmod 755 -R /export

l 切換到realtime用戶下

su realtime

5.3 Kafka集羣部署

5.3.一、下載安裝包

http://kafka.apache.org/downloads.html

在linux中使用wget命令下載安裝包

wget http://mirrors.hust.edu.cn/apache/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz

5.3.二、解壓安裝包

tar -zxvf /export/software/kafka_2.11-0.8.2.2.tgz -C /export/servers/

cd /export/servers/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.三、修改配置文件

cp /export/servers/kafka/config/server.properties

/export/servers/kafka/config/server.properties.bak

vi /export/servers/kafka/config/server.properties

輸入如下內容：

5.3.四、分發安裝包

scp -r /export/servers/kafka_2.11-0.8.2.2 kafka02:/export/servers

而後分別在各機器上建立軟連

cd /export/servers/

ln -s kafka_2.11-0.8.2.2 kafka

5.3.五、再次修改配置文件（重要）

依次修改各服務器上配置文件的的broker.id，分別是0,1,2不得重複。

5.3.六、啓動集羣

首先啓動zookeeper集羣

zkServer.sh start

依次在各節點上啓動kafka

bin/kafka-server-start.sh config/server.properties

集羣啓動後會卡主頁面，須要再開一個窗口調用jps指令查看kafka是否已經啓動

5.四、Kafka經常使用操做命令

l 查看當前服務器中的全部topic

bin/kafka-topics.sh --list --zookeeper zk01:2181

l 建立topic

./kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 1 --partitions 3 --topic first

l 刪除topic

sh bin/kafka-topics.sh --delete --zookeeper zk01:2181 --topic test

須要server.properties中設置delete.topic.enable=true不然只是標記刪除或者直接重啓。

l 經過shell命令發送消息

kafka-console-producer.sh --broker-list kafka01:9092 --topic itheima

l 經過shell消費消息

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

l 查看消費位置

sh kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zk01:2181 --group testGroup

l 查看某個Topic的詳情

sh kafka-topics.sh --topic test --describe --zookeeper zk01:2181

六、Kafka生產者Java API

七、Kafka消費者Java API

StreamMessage：Java 數據流消息，用標準流操做來順序的填充和讀取。

MapMessage：一個Map類型的消息；名稱爲 string 類型，而值爲 Java 的基本類型。

TextMessage：普通字符串消息，包含一個String。

ObjectMessage：對象消息，包含一個可序列化的Java 對象

BytesMessage：二進制數組消息，包含一個byte[]。

XMLMessage: 一個XML類型的消息。

最經常使用的是TextMessage和ObjectMessage。

Kafka負載均衡、Kafka自定義Partition、Kafk文件存儲機制介紹

名稱：

Kafka技術加強

注：請先學習Kafka基礎

目標：

經過本模塊的學習，可以掌握Kafka的負載均衡、Producer生產數據、Kafka文件存儲機制、Kafka自定義partition

課程大綱：

一、 Kafka總體結構圖

二、 Consumer與topic關係

三、 Kafka Producer消息分發

四、 Consumer 的負載均衡

五、 Kafka文件存儲機制

內容

一、Kafka總體結構圖

Kafka名詞解釋和工做方式

l Producer ：消息生產者，就是向kafka broker發消息的客戶端。

l Consumer ：消息消費者，向kafka broker取消息的客戶端

l Topic ：我們能夠理解爲一個隊列。目標發送的目的地，這是一個邏輯上的概念，落到磁盤上是一個partition的目錄。

partition的目錄中有多個segment組合(index,log)；一個Topic對應多個partition[0,1,2,3]，一個partition對應多個segment組合。一個segment有默認的大小是1G。每一個partition能夠設置多個副本(replication-factor 1),會從全部的副本中選取一個leader出來。全部讀寫操做都是經過leader來進行的。

特別強調，和mysql中主從有區別，mysql作主從是爲了讀寫分離，在kafka中讀寫操做都是leader。

l Consumer Group （CG）：這是kafka用來實現一個topic消息的廣播（發給全部的consumer）和單播（發給任意一個consumer）的手段。一個topic能夠有多個CG。topic的消息會複製（不是真的複製，是概念上的）到全部的CG，但每一個partion只會把消息發給該CG中的一個consumer。若是須要實現廣播，只要每一個consumer有一個獨立的CG就能夠了。要實現單播只要全部的consumer在同一個CG。用CG還能夠將consumer進行自由的分組而不須要屢次發送消息到不一樣的topic。

l Broker ：一臺kafka服務器就是一個broker。一個集羣由多個broker組成。一個broker能夠容納多個topic。

l Partition：爲了實現擴展性，一個很是大的topic能夠分佈到多個broker（即服務器）上，一個topic能夠分爲多個partition，每一個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的總體（多個partition間）的順序。

l Offset：kafka的存儲文件都是按照offset.kafka來命名，用offset作名字的好處是方便查找。例如你想找位於2049的位置，只要找到2048.kafka的文件便可。固然the first offset就是00000000000.kafka

二、Consumer與topic關係

本質上kafka只支持Topic；

l 每一個group中能夠有多個consumer，每一個consumer屬於一個consumer group；

一般狀況下，一個group中會包含多個consumer，這樣不只能夠提升topic中消息的併發消費能力，並且還能提升"故障容錯"性，若是group中的某個consumer失效那麼其消費的partitions將會有其餘consumer自動接管。

l 對於Topic中的一條特定的消息，只會被訂閱此Topic的每一個group中的其中一個consumer消費，此消息不會發送給一個group的多個consumer；

那麼一個group中全部的consumer將會交錯的消費整個Topic，每一個group中consumer消息消費互相獨立，咱們能夠認爲一個group是一個"訂閱"者。

l 在kafka中,一個partition中的消息只會被group中的一個consumer消費(同一時刻)；

一個Topic中的每一個partions，只會被一個"訂閱者"中的一個consumer消費，不過一個consumer能夠同時消費多個partitions中的消息。

l kafka的設計原理決定,對於一個topic，同一個group中不能有多於partitions個數的consumer同時消費，不然將意味着某些consumer將沒法獲得消息。

kafka只能保證一個partition中的消息被某個consumer消費時是順序的；事實上，從Topic角度來講,當有多個partitions時,消息仍不是全局有序的。

三、Kafka消息的分發

Producer客戶端負責消息的分發

l kafka集羣中的任何一個broker均可以向producer提供metadata信息,這些metadata中包含"集羣中存活的servers列表"/"partitions leader列表"等信息；

l 當producer獲取到metadata信息以後, producer將會和Topic下全部partition leader保持socket鏈接；

l 消息由producer直接經過socket發送到broker，中間不會通過任何"路由層"，事實上，消息被路由到哪一個partition上由producer客戶端決定；

好比能夠採用"random""key-hash""輪詢"等,若是一個topic中有多個partitions,那麼在producer端實現"消息均衡分發"是必要的。

l 在producer端的配置文件中,開發者能夠指定partition路由的方式。

Producer消息發送的應答機制

設置發送數據是否須要服務端的反饋,有三個值0,1,-1

0: producer不會等待broker發送ack

1: 當leader接收到消息以後發送ack

-1: 當全部的follower都同步消息成功後發送ack

request.required.acks=0

4、Consumer的負載均衡

當一個group中,有consumer加入或者離開時,會觸發partitions均衡.均衡的最終目的,是提高topic的併發消費能力，步驟以下：

一、假如topic1,具備以下partitions: P0,P1,P2,P3

二、加入group中,有以下consumer: C1,C2

三、首先根據partition索引號對partitions排序: P0,P1,P2,P3

四、根據consumer.id排序: C0,C1

五、計算倍數: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)

六、而後依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

五、kafka文件存儲機制

5.一、Kafka文件存儲基本結構

l 在Kafka文件存儲中，同一個topic下有多個不一樣partition，每一個partition爲一個目錄，partiton命名規則爲topic名稱+有序序號，第一個partiton序號從0開始，序號最大值爲partitions數量減1。

l 每一個partion(目錄)至關於一個巨型文件被平均分配到多個大小相等segment(段)數據文件中。但每一個段segment file消息數量不必定相等，這種特性方便old segment file快速被刪除。默認保留7天的數據。

l 每一個partiton只須要支持順序讀寫就好了，segment文件生命週期由服務端配置參數決定。（何時建立，何時刪除）

數據有序的討論？

一個partition的數據是不是有序的？間隔性有序，不連續

針對一個topic裏面的數據，只能作到partition內部有序，不能作到全局有序。

特別加入消費者的場景後，如何保證消費者消費的數據全局有序的？僞命題。

只有一種狀況下才能保證全局有序？就是隻有一個partition。

5.二、Kafka Partition Segment

l Segment file組成：由2大部分組成，分別爲index file和data file，此2個文件一一對應，成對出現，後綴".index"和「.log」分別表示爲segment索引文件、數據文件。

l Segment文件命名規則：partion全局的第一個segment從0開始，後續每一個segment文件名爲上一個segment文件最後一條消息的offset值。數值最大爲64位long大小，19位數字字符長度，沒有數字用0填充。

l 索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。

3，497：當前log文件中的第幾條信息，存放在磁盤上的那個地方

上述圖中索引文件存儲大量元數據，數據文件存儲大量消息，索引文件中元數據指向對應數據文件中message的物理偏移地址。

其中以索引文件中元數據3,497爲例，依次在數據文件中表示第3個message(在全局partiton表示第368772個message)、以及該消息的物理偏移地址爲497。

l segment data file由許多message組成， qq物理結構以下：

關鍵字	解釋說明
8 byte offset	在parition(分區)內的每條消息都有一個有序的id號，這個id號被稱爲偏移(offset),它能夠惟一肯定每條消息在parition(分區)內的位置。即offset表示partiion的第多少message
4 byte message size	message大小
4 byte CRC32	用crc32校驗message
1 byte 「magic"	表示本次發佈Kafka服務程序協議版本號
1 byte 「attributes"	表示爲獨立版本、或標識壓縮類型、或編碼類型。
4 byte key length	表示key的長度,當key爲-1時，K byte key字段不填
K byte key	可選
value bytes payload	表示實際消息數據。