第三單元分布式消息系統(tǒng)Kafka掌握Kafka高可用集群的安裝與配置掌握Kafka與Flume的整合應(yīng)用熟悉Kafka各組件的作用熟悉Kafka中生產(chǎn)者與消費者的使用熟悉Kafka集群管理工具的使用3.1什么是Kafka

Kafka是一個用于發(fā)布（寫）和訂閱（讀）消息的分布式消息系統(tǒng)。它以流的形式從消息源實時獲取數(shù)據(jù)，持久地存儲數(shù)據(jù)，并能實時地響應(yīng)訂閱請求，根據(jù)訂閱請求將消息路由到對應(yīng)的目的地。Kafka以多臺服務(wù)器組成一個集群，集群中的每臺服務(wù)器被稱之為Broker（代理），集群中的Broker數(shù)量越多，集群對消息的吞吐率越高；Kafka以ZooKeeper作為監(jiān)管系統(tǒng)，如果集群中的某個Broker發(fā)生故障，其它Broker會接管其工作，確保集群能連續(xù)工作。3.2Kafka的主要組件

Kafka的組件分別有：(1)Producer：消息的生產(chǎn)者，負責(zé)發(fā)布消息到Broker；(2)Consumer：消息的消費者，負責(zé)從Broker中讀取消息；(3)Topic：消息的主題，也稱為消息類別。消息被組織并持久地存儲在Topic，Topic類似于文件系統(tǒng)中的文件夾，消息就是文件夾中的文件。Kafka集群可以同時處理多個Topic的分發(fā)。(4)Partition：Topic在物理上的分組，一個Topic可以分為多個Partition。(5)Segment：每個Partition相當(dāng)于一個巨型文件，被平均分配到多個大小相等Segment(段)數(shù)據(jù)文件中。一個Topic可以有零到多個發(fā)布消息的Producer，也可以有零到多個訂閱消息的Consumer。kafka根據(jù)時間策略刪除消息。一個Topic會被分散于不同Broker上，應(yīng)用程序可以同時從多個Broker讀取和寫入數(shù)據(jù)。具有相同消息密鑰的消息會被寫入同一個Partition中，并且Kafka對給定TopicPartition的消費者，將始終以與寫入消息時的完全相同順序讀取該Partition中的消息。3.3安裝與配置Kafka高可用集群模式

在瀏覽器的地址欄中輸入Kafka的官方地址“/downloads”，在打開的頁面中，單擊“Scala2.13-kafka_2.13-2.8.0.tgz(asc,sha512)”處的鏈接，跳轉(zhuǎn)到下載頁面。在跳轉(zhuǎn)后的新頁面中，復(fù)制第一個下載鏈接地址。進入kafka的配置目錄“config”，打開其目錄下的perties文件，對照以下代碼進行修改。#listeners=PLAINTEXT://:9092listeners=PLAINTEXT://node1的主機名:9092......#advertised.listeners=PLAINTEXT://:9092advertised.listeners=PLAINTEXT://node1的主機名:9092#Acommaseparatedlistofdirectoriesunderwhichtostorelogfiles#log.dirs=/tmp/kafka-logslog.dirs=/usr/app/kafka_2.13-2.8.0/logs......num.partitions=3......#zookeeper.connect=localhost:2181zookeeper.connect=node1的主機名:2181,node2的主機名:2181,node3的主機名:2181將配置好的Kafka復(fù)制到節(jié)點node2、node3。[root@node1app]#scp-rkafka_2.13-2.8.0/root@node2:/usr/app/[root@node1app]#scp-rkafka_2.13-2.8.0/root@node3:/usr/app/復(fù)制后，分別修改節(jié)點node2、node3的perties文件，將主機名改為各自節(jié)點自己的主機名。在所有節(jié)點上，修改環(huán)境變量。在打開的文件中，追加寫入。exportKAFKA_HOME=/usr/app/kafka_2.13-2.8.0exportPATH=$PATH:$KAFKA_HOME/bin執(zhí)行命令，使修改生效。source/etc/profile3.4啟動Kafka集群

在所有節(jié)點上啟動ZooKeeper。zkServer.shstart在所有節(jié)點上執(zhí)行下列命令，啟動Kafka。kafka-server-start.sh-daemon/usr/app/kafka_2.13-2.8.0/config/perties執(zhí)行以下命令，檢測啟動狀況。jps在所有節(jié)點的返回信息中，會顯示有以下兩個進程名稱，則表示Kafka集群啟動成功。8898Kafka8043QuorumPeerMain如果要停止Kafka集群，在所有節(jié)點上執(zhí)行下列命令。kafka-server-stop.sh-daemon/usr/app/kafka_2.13-2.8.0/config/perties由于啟動與退出kafka的命令內(nèi)容過長，可以參照書中3.5節(jié)內(nèi)容，通過分別編寫kafka啟動、退出的腳本，來簡化命令。3.5測試Kafka

1.創(chuàng)建Topic

在使用Kafka發(fā)送消息前，必須要存在一個Topic。例如：在主節(jié)點node1上，執(zhí)行以下命令，創(chuàng)建一個名稱為“test”，帶有兩個partition，兩個副本的Topic。[root@node1~]#kafka-topics.sh--create--zookeepernode1主機名:2181,node2主機名:2181,node3主機名:2181--replication-factor2--partitions2--topic

test查看Topic（test）的狀態(tài)。[root@node1~]#kafka-topics.sh--describe--zookeepernode1主機名:2181,node2主機名:2181,node3主機名:2181--topictest2.創(chuàng)建Producer（數(shù)據(jù)的生產(chǎn)者）例如：在主節(jié)點node1上執(zhí)行以下命令，從Kafka集群的broker服務(wù)器清單中選擇node1節(jié)點，做為前面已創(chuàng)建的Topic（test）消息的Producer。[root@node1~]#kafka-console-producer.sh--broker-listnode1主機名:9092--topic

test

創(chuàng)建成功后，node1節(jié)點會處于等待輸入的狀態(tài)。3.創(chuàng)建Consumer（數(shù)據(jù)的消費者）在集群中的其它任意一從節(jié)點上，以節(jié)點node2為例，執(zhí)行以下命令。從Kafka集群已連接的服務(wù)器中選擇node2和node3節(jié)點做為前面已創(chuàng)建的Topic（test）消息的Consumer。[root@node2~]#kafka-console-consumer.sh--bootstrap-servernode2主機名:9092,node3主機名:9092--topictest--from-beginning4.進行消息的發(fā)送與接收在主節(jié)點node1上，輸入測試數(shù)據(jù)，并回車。在從節(jié)點node2上，會自動接收來至主節(jié)點node1發(fā)送的消息。5.消息的刪除Kafka集群上的消息，默認會存放一周，一周后，則會被自動刪除。如果要手動立即刪除，以前面剛剛創(chuàng)建的test

topic為例，可在集群中的任意一節(jié)點上，執(zhí)行以下命令。[root@node1~]#kafka-topics.sh--delete--bootstrap-servernode1主機名:9092--topic

test3.6Kafka集群管理工具kafka-manager 3.6.1 kafka-manager介紹

kafka-manager是一款用于kafka集群管理的Web類軟件工具，提供了便利的菜單操作方式來管理kafka集群，可以提高kafka集群管理的工作效率。kafka-manager是yahoo在github上的一個開源項目（項目地址“/yahoo/CMAK”，截止2021年8月）。kafka-manager提供的有但不限于以下功能：(1)管理多個Cluster（簇），一個kafka集群在邏輯上可以被劃分成多個Cluster；(2)檢查組件狀態(tài)（Topic，Consumer，Broker，Replication，Partition）；(3)創(chuàng)建、查詢、更新、刪除Topic。kafka-manager的安裝請參照書中3.8.1節(jié)3.6.2 kafka-manager的使用 在kafka-manager啟動成功后，可以在瀏覽器中訪問“http://node1:9090/”，打開kafka-manager的WEB頁面。單擊頁面中的“Cluster”下拉列表菜單中的“AddCluster”菜單項。在刷新后的頁面中，輸入集群的名字和Zookeeper集群各節(jié)點地址（格式為：節(jié)點主機名:2181端口號,節(jié)點主機名:2181端口號，節(jié)點之間用英文的逗號間隔）等，選擇最高的Kafka版本，最后，在最下方，單擊“Save”按鈕保存。在“ClusterInformation”頁面中，單擊菜單欄上的“Topic”下拉列表菜單中的“Create”菜單項，頁面跳轉(zhuǎn)到“CreateTopic”頁面。在“CreateTopic”頁面，填入主題的名稱、分區(qū)數(shù)量、副本數(shù)量后，單擊“Create”按鈕，完成Topic的創(chuàng)建。在打開的“Test”Topic頁面上，可以看到右邊有關(guān)于Topic的功能操作區(qū)（增、刪、改等按鈕），在頁面的下方顯示有當(dāng)前集群服務(wù)器的相關(guān)狀態(tài)信息。3.7日志采集項目，整合Flume+Kafka案例 3.7.1 創(chuàng)建模擬日志數(shù)據(jù)源項目

在eclipse工具中創(chuàng)建Maven項目，pom.xml為項目的配置文件。創(chuàng)建好的項目結(jié)構(gòu)如下：在項目中創(chuàng)建“OrdersInfo”和“LoggerProducer”兩個Java類，以及perties日志配置文件。OrdersInfo.java文件：用于定義日志數(shù)據(jù)的結(jié)構(gòu)和日志內(nèi)容，LoggerProducer.java文件：作為項目的程序運行入口，調(diào)用OrdersInfo.java類。perties文件：負責(zé)日志的配置。將編寫好代碼的模擬日志數(shù)據(jù)源項目，編譯成jar文件，提交到集群主節(jié)點node1中。3.7.2 編寫采集方案，實現(xiàn)Flume+Kafka整合

#設(shè)置sourcechannelsink......a1.sources.r1.channels=c1#指定source的管道a1.sources.r1.type=TAILDIR#指定source類型a1.sources.r1.positionFile=/usr/test/flume-test/flume-kafka/record/taildir_position.jsona1.sources.r1.filegroups=f1#指定被監(jiān)視的文件a1.sources.r1.filegroups.f1=/usr/test/flume-test/flume-kafka/orderLogs/orderinfo.loga1.channels.c1.type=memory#指定channel為memory類型a1.sinks.k1.channel=c1#指定sink的管道#指定sink類型為kafka，注意KafkaSink字母的大小，第一個K大寫，第一個S大寫a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink#將topic的值指定為前面創(chuàng)建好的kafka中的topic名稱a1.sinks.k1.kafka.topic=kafka_testa1.sinks.k1.kafka.bootstrap.servers=node1主機名:9092,node2主機名:9092,node3主機名:9092a1.sinks.k1.kafka.flumeBatchSize=20ducer.acks=13.7.3 操作部分

在主節(jié)點node1的“flume-kafka”目錄下，啟動Flume采集方案:[root@node1flume-kafka]#flume-ngagent-cconf/-fflume-kafka.conf-na1-Dflume.root.logger=INFO,console另開一個主節(jié)點node1的終端