1、Kafka使用背景

成都創(chuàng)新互聯(lián)是一家集網(wǎng)站建設(shè),沅江企業(yè)網(wǎng)站建設(shè),沅江品牌網(wǎng)站建設(shè),網(wǎng)站定制,沅江網(wǎng)站建設(shè)報價,網(wǎng)絡(luò)營銷,網(wǎng)絡(luò)優(yōu)化,沅江網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強企業(yè)競爭力?？沙浞譂M足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時我們時刻保持專業(yè)、時尚、前沿，時刻以成就客戶成長自我，堅持不斷學(xué)習(xí)、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實用型網(wǎng)站。

在我們大量使用分布式數(shù)據(jù)庫、分布式計算集群的時候，是否會遇到這樣的一些問題：

1. 我們想分析下用戶行為（pageviews），以便我們設(shè)計出更好的廣告位

2. 我想對用戶的搜索關(guān)鍵詞進行統(tǒng)計，分析出當(dāng)前的流行趨勢

3. 有些數(shù)據(jù)，存儲數(shù)據(jù)庫浪費，直接存儲硬盤效率又低

這些場景都有一個共同點：

數(shù)據(jù)是由上游模塊產(chǎn)生，上游模塊，使用上游模塊的數(shù)據(jù)計算、統(tǒng)計、分析，這個時候就可以使用消息系統(tǒng)，尤其是分布式消息系統(tǒng)！

2、Kafka的定義

What is Kafka：它是一個分布式消息系統(tǒng)，由linkedin使用scala編寫，用作LinkedIn的活動流（Activity Stream）和運營數(shù)據(jù)處理管道（Pipeline）的基礎(chǔ)。具有高水平擴展和高吞吐量。

3、Kafka和其他主流分布式消息系統(tǒng)的對比

定義解釋：

1、Java 和 scala都是運行在JVM上的語言。

2、erlang和最近比較火的和go語言一樣是從代碼級別就支持高并發(fā)的一種語言，所以RabbitMQ天生就有很高的并發(fā)性能，但是 有RabbitMQ嚴格按照AMQP進行實現(xiàn)，受到了很多限制。kafka的設(shè)計目標是高吞吐量，所以kafka自己設(shè)計了一套高性能但是不通用的協(xié)議，他也是仿照AMQP（ Advanced Message Queuing Protocol ? 高級消息隊列協(xié)議）設(shè)計的。

3、事務(wù)的概念：在數(shù)據(jù)庫中，多個操作一起提交，要么操作全部成功，要么全部失敗。舉個例子， 在轉(zhuǎn)賬的時候付款和收款，就是一個事物的例子，你給一個人轉(zhuǎn)賬，你轉(zhuǎn)成功，并且對方正常行收到款項后，這個操作才算成功，有一方失敗，那么這個操作就是失敗的。

對應(yīng)消在息隊列中，就是多條消息一起發(fā)送，要么全部成功，要么全部失敗。3個中只有ActiveMQ支持，這個是因為，RabbitMQ和Kafka為了更高的性能，而放棄了對事務(wù)的支持 。

4、集群：多臺服務(wù)器組成的整體叫做集群，這個整體對生產(chǎn)者和消費者來說，是透明的。其實對消費系統(tǒng)組成的集群添加一臺服務(wù)器減少一臺服務(wù)器對生產(chǎn)者和消費者都是無感之的。

5、負載均衡，對消息系統(tǒng)來說負載均衡是大量的生產(chǎn)者和消費者向消息系統(tǒng)發(fā)出請求消息，系統(tǒng)必須均衡這些請求使得每一臺服務(wù)器的請求達到平衡，而不是大量的請求，落到某一臺或幾臺，使得這幾臺服務(wù)器高負荷或超負荷工作，嚴重情況下會停止服務(wù)或宕機。

6、動態(tài)擴容是很多公司要求的技術(shù)之一，不支持動態(tài)擴容就意味著停止服務(wù)，這對很多公司來說是不可以接受的。

注：

阿里巴巴的Metal,RocketMQ都有Kafka的影子，他們要么改造了Kafka或者借鑒了Kafka，最后Kafka的動態(tài)擴容是通過Zookeeper來實現(xiàn)的。

Zookeeper是一種在分布式系統(tǒng)中被廣泛用來作為：分布式狀態(tài)管理、分布式協(xié)調(diào)管理、分布式配置管理、和分布式鎖服務(wù)的集群。kafka增加和減少服務(wù)器都會在Zookeeper節(jié)點上觸發(fā)相應(yīng)的事件kafka系統(tǒng)會捕獲這些事件，進行新一輪的負載均衡，客戶端也會捕獲這些事件來進行新一輪的處理。

二、Kafka相關(guān)概念

1、 AMQP協(xié)議

Advanced Message Queuing Protocol （高級消息隊列協(xié)議）

The Advanced Message Queuing Protocol (AMQP)：是一個標準開放的應(yīng)用層的消息中間件（Message Oriented Middleware）協(xié)議。AMQP定義了通過網(wǎng)絡(luò)發(fā)送的字節(jié)流的數(shù)據(jù)格式。因此兼容性非常好，任何實現(xiàn)AMQP協(xié)議的程序都可以和與AMQP協(xié)議兼容的其他程序交互，可以很容易做到跨語言，跨平臺。

上面說的3種比較流行的消息隊列協(xié)議，要么支持AMQP協(xié)議，要么借鑒了AMQP協(xié)議的思想進行了開發(fā)、實現(xiàn)、設(shè)計。

2、 一些基本的概念

1、消費者：（Consumer）：從消息隊列中請求消息的客戶端應(yīng)用程序

2、生產(chǎn)者：（Producer） ?：向broker發(fā)布消息的應(yīng)用程序

3、AMQP服務(wù)端（broker）：用來接收生產(chǎn)者發(fā)送的消息并將這些消息路由給服務(wù)器中的隊列，便于fafka將生產(chǎn)者發(fā)送的消息，動態(tài)的添加到磁盤并給每一條消息一個偏移量，所以對于kafka一個broker就是一個應(yīng)用程序的實例

kafka支持的客戶端語言

：Kafka客戶端支持當(dāng)前大部分主流語言，包括：C、C++、Erlang、Java、.net、perl、PHP、Python、Ruby、Go、Javascript

可以使用以上任何一種語言和kafka服務(wù)器進行通信（即辨析自己的consumer從kafka集群訂閱消息也可以自己寫producer程序）

3、Kafka架構(gòu)

生產(chǎn)者生產(chǎn)消息、kafka集群、消費者獲取消息這樣一種架構(gòu)，如下圖：

kafka集群中的消息，是通過Topic（主題）來進行組織的，如下圖：

一些基本的概念：

1、主題（Topic）：一個主題類似新聞中的體育、娛樂、教育等分類概念，在實際工程中通常一個業(yè)務(wù)一個主題。

2、分區(qū)（Partition）：一個Topic中的消息數(shù)據(jù)按照多個分區(qū)組織，分區(qū)是kafka消息隊列組織的最小單位，一個分區(qū)可以看作是一個FIFO（ First Input First Output的縮寫，先入先出隊列）的隊列。

kafka分區(qū)是提高kafka性能的關(guān)鍵所在，當(dāng)你發(fā)現(xiàn)你的集群性能不高時，常用手段就是增加Topic的分區(qū)，分區(qū)里面的消息是按照從新到老的順序進行組織，消費者從隊列頭訂閱消息，生產(chǎn)者從隊列尾添加消息。

工作圖：

備份（Replication）：為了保證分布式可靠性，kafka0.8開始對每個分區(qū)的數(shù)據(jù)進行備份（不同的Broker上），防止其中一個Broker宕機造成分區(qū)上的數(shù)據(jù)不可用。

kafka0.7是一個很大的改變：1、增加了備份2、增加了控制借點概念，增加了集群領(lǐng)導(dǎo)者選舉 。

三、Zookeeper(動物園)集群搭建

kafka集群是把狀態(tài)保存在Zookeeper中的，首先要搭建Zookeeper集群。

1、軟件環(huán)境(3臺服務(wù)器-測試,一般是奇數(shù)臺服務(wù)器)

vim? /etc/hosts(3臺服務(wù)器都要寫)

192.168.11.128? server1

192.168.11.129? server2

192.168.11.130? server3

1、Linux服務(wù)器一臺、三臺、五臺、（2*n+1），Zookeeper集群的工作是超過半數(shù)才能對外提供服務(wù)，3臺中超過兩臺超過半數(shù)，允許1臺掛掉 ，是否可以用偶數(shù)，其實沒必要。

如果有四臺那么掛掉一臺還剩下三臺服務(wù)器，如果在掛掉一個就不行了，這里記住是超過半數(shù)。

2、Java jdk1.8 zookeeper是用java寫的所以他的需要JAVA環(huán)境，java是運行在java虛擬機上的

3、Zookeeper的穩(wěn)定版本Zookeeper 3.4.14版本

-------------------------------------------------------------------------------------------------------------------------------------------------

2、配置安裝zookeeper

下面的操作是: 3臺服務(wù)器統(tǒng)一操作

1、安裝java（我這里采用jdk安裝）

先準備jdk的包，解壓帶/usr/local下

創(chuàng)建軟鏈接

寫java環(huán)境變量

vim? ?/etc/profile.d/aa.sh

加載環(huán)境變量

source? /etc/profile(加載全部環(huán)境變量)? ?或者? source /etc/profile.d/aa.sh(加載這一個環(huán)境變量)

查看java有沒有安裝成功

如上圖所示，表示java環(huán)境已經(jīng)部署成功

------------------------------------------------------------------------------------------------------------------------

用yum安裝java

yum? list? ?java*? ? &&? ?yum? -y? install? java

------------------------------------------------------------------------------------------------------------------------

2、下載Zookeeper（3臺服務(wù)器統(tǒng)一操作）

首先要注意在生產(chǎn)環(huán)境中目錄結(jié)構(gòu)要定義好，防止在項目過多的時候找不到所需的項目

#目錄統(tǒng)一放在/opt下面

#首先創(chuàng)建Zookeeper項目目錄

cd? /opt

mkdir? ?zookeeper? ? ? ? ?//項目目錄

mkdir? ?zookeeper/zkdata? ? ? ? //存放快照日志

mkdir? ?zookeeper/zkdatalog? ? ? ?//存放事物日志

下載Zookeeper

#下載軟件

cd? ?/opt/zookeeper/

wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

解壓軟件

tar? ?xf? ??zookeeper-3.4.14.tar.gz

3、修改配置文件

進入到解壓好的目錄里面的conf目錄中，查看

#zoo_sample.cfg ?這個文件是官方給我們的zookeeper的樣板文件，給他復(fù)制一份命名為zoo.cfg，zoo.cfg是官方指定的文件命名規(guī)則。

3臺服務(wù)器的配置文件

vim? ? zoo.cfg

上面定義了dataDir和clientPort這2行就要注釋，否則后面起集群會報錯。

#server.1 這個1是服務(wù)器的標識也可以是其他的數(shù)字， 表示這個是第幾號服務(wù)器，用來標識服務(wù)器，這個標識要寫到快照目錄下面myid文件里

#192.168.11.139為集群里的IP地址，第一個端口是master和slave之間的通信端口，默認是2888，第二個端口是leader選舉的端口，集群剛啟動的時候選舉或者leader掛掉之后進行新的選舉的端口默認是3888

配置文件解釋：

#tickTime：

這個時間是作為 Zookeeper 服務(wù)器之間或客戶端與服務(wù)器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發(fā)送一個心跳。

#initLimit：

這個配置項是用來配置 Zookeeper 接受客戶端（這里所說的客戶端不是用戶連接 Zookeeper 服務(wù)器的客戶端，而是 Zookeeper 服務(wù)器集群中連接到 Leader 的 Follower 服務(wù)器）初始化連接時最長能忍受多少個心跳時間間隔數(shù)。當(dāng)已經(jīng)超過 5個心跳的時間（也就是 tickTime）長度后 Zookeeper 服務(wù)器還沒有收到客戶端的返回信息，那么表明這個客戶端連接失敗?？偟臅r間長度就是 5*2000=10 秒

#syncLimit：

這個配置項標識 Leader 與Follower 之間發(fā)送消息，請求和應(yīng)答時間長度，最長不能超過多少個 tickTime 的時間長度，總的時間長度就是5*2000=10秒

#dataDir：

快照日志的存儲路徑

#dataLogDir：

事物日志的存儲路徑，如果不配置這個那么事物日志會默認存儲到dataDir制定的目錄，這樣會嚴重影響zk的性能，當(dāng)zk吞吐量較大的時候，產(chǎn)生的事物日志、快照日志太多

#clientPort：

這個端口就是客戶端連接 Zookeeper 服務(wù)器的端口，Zookeeper 會監(jiān)聽這個端口，接受客戶端的訪問請求。修改他的端口改大點

創(chuàng)建myid文件（每一臺的都不一樣）

#server1

echo "1" > /opt/zookeeper/zkdata/myid

#server2

echo "2" > /opt/zookeeper/zkdata/myid

#server3

echo "3" > /opt/zookeeper/zkdata/myid

4、重要配置說明

1、myid文件和server.myid ?在快照目錄下存放的標識本臺服務(wù)器的文件，他是整個zk集群用來發(fā)現(xiàn)彼此的一個重要標識。

2、zoo.cfg 文件是zookeeper配置文件 在conf目錄里。

3、log4j.properties文件是zk的日志輸出文件 在conf目錄里用java寫的程序基本上有個共同點日志都用log4j，來進行管理。

4、zkEnv.sh和zkServer.sh文件

zkServer.sh ?主的管理程序文件

zkEnv.sh ? 是主要配置，zookeeper集群啟動時配置環(huán)境變量的文件

5、還有一個需要注意

ZooKeeper serverwill not remove old snapshots and log files when using the default configuration (see autopurge below), this is the responsibility of the operator

zookeeper不會主動的清除舊的快照和日志文件，這個是操作者的責(zé)任。

但是可以通過命令去定期的清理

#以上這個腳本定義了刪除對應(yīng)兩個目錄中的文件，保留最新的66個文件，可以將他寫到crontab中，設(shè)置為每天凌晨2點執(zhí)行一次就可以了。

5、啟動服務(wù)并查看

1、啟動服務(wù)

#進去到Zookeeper的bin目錄下

cd ? /opt/zookeeper/zookeeper-3.4.14/bin

#啟動服務(wù)(3臺都需要操作)

2、檢查服務(wù)狀態(tài)

#檢查服務(wù)器狀態(tài)（會有一個leader和2個follower）

zk集群一般只有一個leader，多個follower，主一般是相應(yīng)客戶端的讀寫請求，而從主同步數(shù)據(jù)，當(dāng)主掛掉之后就會從follower里投票選舉一個leader出來。

可以用jps查看zk的進程，這是整個工程的main

#執(zhí)行命令jps

四，kafka集群搭建

1，軟件環(huán)境

1、linux一臺或多臺，大于等于2

2、已經(jīng)搭建好的zookeeper集群

3、軟件版本kafka_2.11-0.9.0.1.tgz

2、創(chuàng)建目錄并下載安裝軟件（3臺服務(wù)器一起操作）

#創(chuàng)建目錄

cd ? /opt

mkdir ? kafka ? #創(chuàng)建項目目錄

cd ? kafka

mkdir ? kafkalogs ? ?#創(chuàng)建kafka消息目錄，主要存放kafka消息

#下載軟件

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz

#解壓軟件

tar -zxvf kafka_2.11-2.2.0.tgz

3，修改配置文件

進入到config目錄

cd /opt/kafka/kafka_2.11-2.2.0/config/

主要關(guān)注：server.properties 這個文件即可，我們可以發(fā)現(xiàn)在目錄下：

有很多文件，這里可以發(fā)現(xiàn)有Zookeeper文件，我們可以根據(jù)Kafka內(nèi)帶的zk集群來啟動，但是建議使用獨立的zk集群

修改配置文件

broker.id=0? #當(dāng)前機器在集群中的唯一標識，和zookeeper的myid性質(zhì)一樣

port=19092 #當(dāng)前kafka對外提供服務(wù)的端口默認是9092

host.name=192.168.7.100 #這個參數(shù)默認是關(guān)閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。

num.network.threads=3 #這個是borker進行網(wǎng)絡(luò)處理的線程數(shù)

num.io.threads=8 #這個是borker進行I/O處理的線程數(shù)

log.dirs=/opt/kafka/kafkalogs/ #消息存放的目錄，這個目錄可以配置為“，”逗號分割的表達式，上面的num.io.threads要大于這個目錄的個數(shù)這個目錄，如果配置多個目錄，新創(chuàng)建的topic他把消息持久化的地方是，當(dāng)前以逗號分割的目錄中，那個分區(qū)數(shù)最少就放那一個

socket.send.buffer.bytes=102400 #發(fā)送緩沖區(qū)buffer大小，數(shù)據(jù)不是一下子就發(fā)送的，先回存儲到緩沖區(qū)了到達一定的大小后在發(fā)送，能提高性能

socket.receive.buffer.bytes=102400 #kafka接收緩沖區(qū)大小，當(dāng)數(shù)據(jù)到達一定大小后在序列化到磁盤

socket.request.max.bytes=104857600 #這個參數(shù)是向kafka請求消息或者向kafka發(fā)送消息的請請求的大數(shù)，這個值不能超過java的堆棧大小

num.partitions=1 #默認的分區(qū)數(shù)，一個topic默認1個分區(qū)數(shù)

log.retention.hours=168 #默認消息的大持久化時間，168小時，7天

message.max.byte=5242880? #消息保存的大值5M

default.replication.factor=2? #kafka保存消息的副本數(shù)，如果一個副本失效了，另一個還可以繼續(xù)提供服務(wù)

replica.fetch.max.bytes=5242880? #取消息的大直接數(shù)

log.segment.bytes=1073741824 #這個參數(shù)是：因為kafka的消息是以追加的形式落地到文件，當(dāng)超過這個值的時候，kafka會新起一個文件

log.retention.check.interval.ms=300000 #每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄查看是否有過期的消息如果有，刪除

log.cleaner.enable=false #是否啟用log壓縮，一般不用啟用，啟用的話可以提高性能

zookeeper.connect=192.168.11.139:12181,192.168.11.140:12181,192.168.11.141:1218 #設(shè)置zookeeper的連接端口

上面是參數(shù)的解釋，實際的修改項為：

配置文件修改結(jié)束

4,啟動kafka集群并測試

1. 啟動服務(wù)

2.檢查服務(wù)是否啟動

#執(zhí)行命令jps

3. 創(chuàng)建Topic驗證是否創(chuàng)建成功

更多請看官方文檔：http://kafka.apache.org/documentation.html

#創(chuàng)建Topic（話題）

#解釋

--replication-factor 2? ?#復(fù)制兩份

--partitions 1 #創(chuàng)建1個分區(qū)

--topic #主題為meinv

'''在一臺服務(wù)器上創(chuàng)建一個發(fā)布者'''

#創(chuàng)建一個broker，發(fā)布者

'''在一臺服務(wù)器上創(chuàng)建一個訂閱者'''

到此，服務(wù)搭建結(jié)束

5、其他說明標注

5.1、日志說明

默認kafka的日志是保存在/opt/kafka/kafka_2.11-2.2.0/logs目錄下的，這里說幾個需要注意的日志

server.log #kafka的運行日志
state-change.log ?#kafka他是用zookeeper來保存狀態(tài)，所以他可能會進行切換，切換的日志就保存在這里

controller.log #kafka選擇一個節(jié)點作為“controller”,當(dāng)發(fā)現(xiàn)有節(jié)點down掉的時候它負責(zé)在游泳分區(qū)的所有節(jié)點中選擇新的leader,這使得Kafka可以批量的高效的管理所有分區(qū)節(jié)點的主從關(guān)系。如果controller down掉了，活著的節(jié)點中的一個會備切換為新的controller.

5.2、上面的大家你完成之后可以登錄zk來查看zk的目錄情況

#使用客戶端進入zk
cd/opt/zookeeper/zookeeper-3.4.14/bin
./zkCli.sh -server 192.168.11.139:12181 ?#默認是不用加’-server‘參數(shù)的因為我們修改了他的端口

#查看目錄情況 執(zhí)行“l(fā)s /”
[zk: 127.0.0.1:12181(CONNECTED) 0] ls /

#顯示結(jié)果：[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch]
'''
上面的顯示結(jié)果中：只有zookeeper是，zookeeper原生的，其他都是Kafka創(chuàng)建的
'''

#標注一個重要的
[zk: 127.0.0.1:12181(CONNECTED) 1] get /brokers/ids/1
{"jmx_port":-1,"timestamp":"1456125963355","endpoints":["PLAINTEXT://192.168.7.100:19092"],"host":"192.168.7.100","version":2,"port":19092}
cZxid = 0x1000001c1
ctime = Mon Feb 22 15:26:03 CST 2016
mZxid = 0x1000001c1
mtime = Mon Feb 22 15:26:03 CST 2016
pZxid = 0x1000001c1
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x152e40aead20016
dataLength = 139
numChildren = 0
[zk: 127.0.0.1:12181(CONNECTED) 2]

#還有一個是查看partion
[zk: 127.0.0.1:12181(CONNECTED) 7] get /brokers/topics/shuaige/partitions/0
null
cZxid = 0x100000029
ctime = Mon Feb 22 10:05:11 CST 2016
mZxid = 0x100000029
mtime = Mon Feb 22 10:05:11 CST 2016
pZxid = 0x10000002a
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
[zk: 127.0.0.1:12181(CONNECTED) 8]

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

分享名稱：kafka集群搭建(消息)-創(chuàng)新互聯(lián)
文章地址：http://jinyejixie.com/article18/dcjcdp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站建設(shè)、移動網(wǎng)站建設(shè)、App設(shè)計、靜態(tài)網(wǎng)站、用戶體驗、網(wǎng)站設(shè)計公司

廣告

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