如何進(jìn)行kafka的安裝和使用-創(chuàng)新互聯(lián)

這篇文章將為大家詳細(xì)講解有關(guān)如何進(jìn)行kafka的安裝和使用，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

創(chuàng)新互聯(lián)公司自2013年起，先為梁溪等服務(wù)建站，梁溪等地企業(yè)，進(jìn)行企業(yè)商務(wù)咨詢服務(wù)。為梁溪企業(yè)網(wǎng)站制作PC+手機(jī)+微官網(wǎng)三網(wǎng)同步一站式服務(wù)解決您的所有建站問(wèn)題。

1.       kafka介紹

1.1.       主要功能

根據(jù)官網(wǎng)的介紹，ApacheKafka?是一個(gè)分布式流媒體平臺(tái)，它主要有3種功能：

　　1：It lets you publish and subscribe to streams of records.發(fā)布和訂閱消息流，這個(gè)功能類似于消息隊(duì)列，這也是kafka歸類為消息隊(duì)列框架的原因

　　2：It lets you store streams of records in a fault-tolerant way.以容錯(cuò)的方式記錄消息流，kafka以文件的方式來(lái)存儲(chǔ)消息流

　　3：It lets you process streams of records as they occur.可以再消息發(fā)布的時(shí)候進(jìn)行處理

1.2.       使用場(chǎng)景

1：Building real-time streaming data pipelines that reliably get data between systems or applications.在系統(tǒng)或應(yīng)用程序之間構(gòu)建可靠的用于傳輸實(shí)時(shí)數(shù)據(jù)的管道，消息隊(duì)列功能

2：Building real-time streaming applications that transform or react to the streams of data。構(gòu)建實(shí)時(shí)的流數(shù)據(jù)處理程序來(lái)變換或處理數(shù)據(jù)流，數(shù)據(jù)處理功能

1.3.       詳細(xì)介紹

Kafka目前主要作為一個(gè)分布式的發(fā)布訂閱式的消息系統(tǒng)使用，下面簡(jiǎn)單介紹一下kafka的基本機(jī)制

1.3.1 消息傳輸流程

　　　　Producer即生產(chǎn)者，向Kafka集群發(fā)送消息，在發(fā)送消息之前，會(huì)對(duì)消息進(jìn)行分類，即Topic，上圖展示了兩個(gè)producer發(fā)送了分類為topic1的消息，另外一個(gè)發(fā)送了topic2的消息。

　　　　Topic即主題，通過(guò)對(duì)消息指定主題可以將消息分類，消費(fèi)者可以只關(guān)注自己需要的Topic中的消息

　　　　Consumer即消費(fèi)者，消費(fèi)者通過(guò)與kafka集群建立長(zhǎng)連接的方式，不斷地從集群中拉取消息，然后可以對(duì)這些消息進(jìn)行處理。

　　　　從上圖中就可以看出同一個(gè)Topic下的消費(fèi)者和生產(chǎn)者的數(shù)量并不是對(duì)應(yīng)的。

1.3.2 kafka服務(wù)器消息存儲(chǔ)策略

 

　　　　談到kafka的存儲(chǔ)，就不得不提到分區(qū)，即partitions，創(chuàng)建一個(gè)topic時(shí)，同時(shí)可以指定分區(qū)數(shù)目，分區(qū)數(shù)越多，其吞吐量也越大，但是需要的資源也越多，同時(shí)也會(huì)導(dǎo)致更高的不可用性，kafka在接收到生產(chǎn)者發(fā)送的消息之后，會(huì)根據(jù)均衡策略將消息存儲(chǔ)到不同的分區(qū)中。

 

　　在每個(gè)分區(qū)中，消息以順序存儲(chǔ)，最晚接收的的消息會(huì)最后被消費(fèi)。

1.3.3 與生產(chǎn)者的交互

 

　　　　生產(chǎn)者在向kafka集群發(fā)送消息的時(shí)候，可以通過(guò)指定分區(qū)來(lái)發(fā)送到指定的分區(qū)中

　　　　也可以通過(guò)指定均衡策略來(lái)將消息發(fā)送到不同的分區(qū)中

　　　　如果不指定，就會(huì)采用默認(rèn)的隨機(jī)均衡策略，將消息隨機(jī)的存儲(chǔ)到不同的分區(qū)中

1.3.4  與消費(fèi)者的交互

  

　　　　在消費(fèi)者消費(fèi)消息時(shí)，kafka使用offset來(lái)記錄當(dāng)前消費(fèi)的位置

　　　　在kafka的設(shè)計(jì)中，可以有多個(gè)不同的group來(lái)同時(shí)消費(fèi)同一個(gè)topic下的消息，如圖，我們有兩個(gè)不同的group同時(shí)消費(fèi)，他們的的消費(fèi)的記錄位置offset各不項(xiàng)目，不互相干擾。

　　　　對(duì)于一個(gè)group而言，消費(fèi)者的數(shù)量不應(yīng)該多余分區(qū)的數(shù)量，因?yàn)樵谝粋€(gè)group中，每個(gè)分區(qū)至多只能綁定到一個(gè)消費(fèi)者上，即一個(gè)消費(fèi)者可以消費(fèi)多個(gè)分區(qū)，一個(gè)分區(qū)只能給一個(gè)消費(fèi)者消費(fèi)

　　　　因此，若一個(gè)group中的消費(fèi)者數(shù)量大于分區(qū)數(shù)量的話，多余的消費(fèi)者將不會(huì)收到任何消息。

2.       Kafka安裝與使用

2.1.       下載

　　你可以在kafka官網(wǎng) http://kafka.apache.org/downloads下載到最新的kafka安裝包，選擇下載二進(jìn)制版本的tgz文件，根據(jù)網(wǎng)絡(luò)狀態(tài)可能需要fq，這里我們選擇的版本是0.11.0.1，目前的最新版

2.2.       安裝

　　Kafka是使用scala編寫的運(yùn)行與jvm虛擬機(jī)上的程序，雖然也可以在windows上使用，但是kafka基本上是運(yùn)行在linux服務(wù)器上，因此我們這里也使用linux來(lái)開始今天的實(shí)戰(zhàn)。

　　首先確保你的機(jī)器上安裝了jdk，kafka需要java運(yùn)行環(huán)境，以前的kafka還需要zookeeper，新版的kafka已經(jīng)內(nèi)置了一個(gè)zookeeper環(huán)境，所以我們可以直接使用

　　說(shuō)是安裝，如果只需要進(jìn)行最簡(jiǎn)單的嘗試的話我們只需要解壓到任意目錄即可，這里我們將kafka壓縮包解壓到/home目錄

2.3.       配置

　　在kafka解壓目錄下下有一個(gè)config的文件夾，里面放置的是我們的配置文件

　　consumer.properites 消費(fèi)者配置，這個(gè)配置文件用于配置于2.5節(jié)中開啟的消費(fèi)者，此處我們使用默認(rèn)的即可

　　producer.properties 生產(chǎn)者配置，這個(gè)配置文件用于配置于2.5節(jié)中開啟的生產(chǎn)者，此處我們使用默認(rèn)的即可

　　server.properties kafka服務(wù)器的配置，此配置文件用來(lái)配置kafka服務(wù)器，目前僅介紹幾個(gè)最基礎(chǔ)的配置

1. broker.id 申明當(dāng)前kafka服務(wù)器在集群中的唯一ID，需配置為integer,并且集群中的每一個(gè)kafka服務(wù)器的id都應(yīng)是唯一的，我們這里采用默認(rèn)配置即可

2. listeners 申明此kafka服務(wù)器需要監(jiān)聽的端口號(hào)，如果是在本機(jī)上跑虛擬機(jī)運(yùn)行可以不用配置本項(xiàng)，默認(rèn)會(huì)使用localhost的地址，如果是在遠(yuǎn)程服務(wù)器上運(yùn)行則必須配置，例如：

　　　　　　　　　　listeners=PLAINTEXT:// 192.168.180.128:9092。并確保服務(wù)器的9092端口能夠訪問(wèn)

　　　　3.zookeeper.connect 申明kafka所連接的zookeeper的地址 ，需配置為zookeeper的地址，由于本次使用的是kafka高版本中自帶zookeeper，使用默認(rèn)配置即可

　　　　　　　　　zookeeper.connect=localhost:2181

2.4.       運(yùn)行

1. 啟動(dòng)zookeeper

cd進(jìn)入kafka解壓目錄，輸入

bin/zookeeper-server-start.sh config/zookeeper.properties &

啟動(dòng)zookeeper成功后會(huì)看到如下的輸出

　　　　
      2.啟動(dòng)kafka

cd進(jìn)入kafka解壓目錄，輸入

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

啟動(dòng)kafka成功后會(huì)看到如下的輸出

 

2.5.       第一個(gè)消息

　　　2.5.1   創(chuàng)建一個(gè)topic

　　　　Kafka通過(guò)topic對(duì)同一類的數(shù)據(jù)進(jìn)行管理，同一類的數(shù)據(jù)使用同一個(gè)topic可以在處理數(shù)據(jù)時(shí)更加的便捷

　　　　在kafka解壓目錄打開終端，輸入

　　　　bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

　　　　創(chuàng)建一個(gè)名為test的topic

 

        　在創(chuàng)建topic后可以通過(guò)輸入

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

   來(lái)查看已經(jīng)創(chuàng)建的topic

　　2.4.2   創(chuàng)建一個(gè)消息消費(fèi)者

　　　在kafka解壓目錄打開終端，輸入(from-beginning每次都是從頭消費(fèi)，不想從頭消費(fèi)可以取消參數(shù))

　　　　bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

　　　可以創(chuàng)建一個(gè)用于消費(fèi)topic為test的消費(fèi)者

 

         消費(fèi)者創(chuàng)建完成之后，因?yàn)檫€沒有發(fā)送任何數(shù)據(jù)，因此這里在執(zhí)行后沒有打印出任何數(shù)據(jù)

         不過(guò)別著急，不要關(guān)閉這個(gè)終端，打開一個(gè)新的終端，接下來(lái)我們創(chuàng)建第一個(gè)消息生產(chǎn)者

　　2.4.3         創(chuàng)建一個(gè)消息生產(chǎn)者

　　　　在kafka解壓目錄打開一個(gè)新的終端，輸入

　　　　bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

　　　　在執(zhí)行完畢后會(huì)進(jìn)入的編輯器頁(yè)面

在發(fā)送完消息之后，可以回到我們的消息消費(fèi)者終端中，可以看到，終端中已經(jīng)打印出了我們剛才發(fā)送的消息

Python偽代碼版本
消費(fèi)者

1. [root@ip-10-1-2-175 sh]# more cus.py

2. import time, json

3. from pykafka import KafkaClient

4. client = KafkaClient(hosts="10.1.2.175:9092") # 可接受多個(gè)Client這是重點(diǎn)

5. topic = client.topics['test']   # 選擇一個(gè)topic

6. # 生成一個(gè)消費(fèi)者

7. balanced_consumer = topic.get_balanced_consumer(consumer_group='goods_group',auto_commit_enable=True,zookeeper_connect='localhost:2181')

9. for message in balanced_consumer:

10.      print message

生產(chǎn)者

1. [root@ip-10-1-2-175 sh]# more prod.py

2. import time, json

3. from pykafka import KafkaClient

4. def pro():

5.         client = KafkaClient(hosts="10.1.2.175:9092")

6.         topic = client.topics['test']     # 選擇一個(gè)topic

7.         producer = topic.get_producer() # 創(chuàng)建一個(gè)生產(chǎn)者

9.         goods_dict = {'option_type':'insert','option_obj':{'goods_name':'goods-1'} }

10.         goods_json = json.dumps(goods_dict)

11.         producer.produce(goods_json) # 生產(chǎn)消息

12.         producer.stop()

13. if __name__ == '__main__':

14.     pro()

啟動(dòng)消費(fèi)者
[root@ip-10-1-2-175 sh]# python cus.py 

啟動(dòng)生產(chǎn)者
[root@ip-10-1-2-175 sh]# python prod.py 


查看消費(fèi)者
[root@ip-10-1-2-175 sh]# python cus.py 

關(guān)于如何進(jìn)行kafka的安裝和使用就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。

新聞名稱：如何進(jìn)行kafka的安裝和使用-創(chuàng)新互聯(lián)
網(wǎng)站網(wǎng)址：http://bm7419.com/article6/cdggig.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站排名、關(guān)鍵詞優(yōu)化、電子商務(wù)、標(biāo)簽優(yōu)化、定制開發(fā)、商城網(wǎng)站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)