kafka集群搭建(消息)

1、Kafka使用背景

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

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

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

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

這些場景都有一個共同點(diǎn)：

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

2、Kafka的定義

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

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

定義解釋：

1、Java 和 scala都是運(yùn)行在JVM上的語言。

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

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

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

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

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

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

注：

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

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

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

1、 AMQP協(xié)議

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

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

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

2、 一些基本的概念

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

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

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

kafka支持的客戶端語言

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

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

3、Kafka架構(gòu)

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

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

一些基本的概念：

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

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

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

工作圖：

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

kafka0.7是一個很大的改變：1、增加了備份2、增加了控制借點(diǎn)概念，增加了集群領(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ù)，其實(shí)沒必要。

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

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

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

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

2、配置安裝zookeeper

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

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

先準(zhǔn)備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)要定義好，防止在項(xiàng)目過多的時(shí)候找不到所需的項(xiàng)目

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

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

cd? /opt

mkdir? ?zookeeper? ? ? ? ?//項(xiàng)目目錄

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、修改配置文件

進(jìn)入到解壓好的目錄里面的conf目錄中，查看

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

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

vim? ? zoo.cfg

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

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

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

配置文件解釋：

#tickTime：

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

#initLimit：

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

#syncLimit：

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

#dataDir：

快照日志的存儲路徑

#dataLogDir：

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

#clientPort：

這個端口就是客戶端連接 Zookeeper 服務(wù)器的端口，Zookeeper 會監(jiān)聽這個端口，接受客戶端的訪問請求。修改他的端口改大點(diǎ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 ?在快照目錄下存放的標(biāo)識本臺服務(wù)器的文件，他是整個zk集群用來發(fā)現(xiàn)彼此的一個重要標(biāo)識。

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

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

4、zkEnv.sh和zkServer.sh文件

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

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

5、還有一個需要注意

ZooKeeper server will 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點(diǎn)執(zhí)行一次就可以了。

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

1、啟動服務(wù)

#進(jìn)去到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的進(jìn)程，這是整個工程的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)建項(xià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，修改配置文件

進(jìn)入到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集群來啟動，但是建議使用獨(dú)立的zk集群

修改配置文件

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

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

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

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

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

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

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

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

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

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

log.retention.hours=168 #默認(rèn)消息的最大持久化時(shí)間，168小時(shí)，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ù)是：因?yàn)閗afka的消息是以追加的形式落地到文件，當(dāng)超過這個值的時(shí)候，kafka會新起一個文件

log.retention.check.interval.ms=300000 #每隔300000毫秒去檢查上面配置的log失效時(shí)間（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ù)的解釋，實(shí)際的修改項(xiàng)為：

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

4,啟動kafka集群并測試

1. 啟動服務(wù)

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

#執(zhí)行命令jps

3. 創(chuàng)建Topic驗(yàn)證是否創(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、其他說明標(biāo)注

5.1、日志說明

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

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

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

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

#使用客戶端進(jìn)入zk
cd/opt/zookeeper/zookeeper-3.4.14/bin
./zkCli.sh -server 192.168.11.139:12181 ?#默認(rèn)是不用加’-server‘參數(shù)的因?yàn)槲覀冃薷牧怂亩丝?/span>

#查看目錄情況 執(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)建的
'''

#標(biāo)注一個重要的
[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]