Kafka集羣搭建

Kafka初識

一、Kafka使用背景html

在咱們大量使用分佈式數據庫、分佈式計算集羣的時候，是否會遇到這樣的一些問題：

咱們想分析下用戶行爲（pageviews），以便咱們設計出更好的廣告位
我想對用戶的搜索關鍵詞進行統計，分析出當前的流行趨勢
有些數據，存儲數據庫浪費，直接存儲硬盤效率又低

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

數據是由上游模塊產生，上游模塊，使用上游模塊的數據計算、統計、分析，這個時候就可使用消息系統，尤爲是分佈式消息系統！

二、Kafka的定義

What is Kafka：它是一個分佈式消息系統，由linkedin使用scala編寫，用做LinkedIn的活動流（Activity Stream）和運營數據處理管道（Pipeline）的基礎。具備高水平擴展和高吞吐量。

三、Kafka和其餘主流分佈式消息系統的對比

定義解釋：

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

二、erlang和最近比較火的和go語言同樣是從代碼級別就支持高併發的一種語言，因此RabbitMQ天生就有很高的併發性能，可是有RabbitMQ嚴格按照AMQP進行實現，受到了不少限制。kafka的設計目標是高吞吐量，因此kafka本身設計了一套高性能可是不通用的協議，他也是仿照AMQP（ Advanced Message Queuing Protocol 高級消息隊列協議）設計的。

三、事物的概念：在數據庫中，多個操做一塊兒提交，要麼操做所有成功，要麼所有失敗。舉個例子，在轉帳的時候付款和收款，就是一個事物的例子，你給一我的轉帳，你轉成功，而且對方正常行收到款項後，這個操做纔算成功，有一方失敗，那麼這個操做就是失敗的。

對應消在息隊列中，就是多條消息一塊兒發送，要麼所有成功，要麼所有失敗。3箇中只有ActiveMQ支持，這個是由於，RabbitMQ和Kafka爲了更高的性能，而放棄了對事物的支持。

四、集羣：多臺服務器組成的總體叫作集羣，這個總體對生產者和消費者來講，是透明的。其實對消費系統組成的集羣添加一臺服務器減小一臺服務器對生產者和消費者都是無感之的。

五、負載均衡，對消息系統來講負載均衡是大量的生產者和消費者向消息系統發出請求消息，系統必須均衡這些請求使得每一臺服務器的請求達到平衡，而不是大量的請求，落到某一臺或幾臺，使得這幾臺服務器高負荷或超負荷工做，嚴重狀況下會中止服務或宕機。

六、動態擴容是不少公司要求的技術之一，不支持動態擴容就意味着中止服務，這對不少公司來講是不能夠接受的。

注：

阿里巴巴的Metal,RocketMQ都有Kafka的影子，他們要麼改造了Kafka或者借鑑了Kafka，最後Kafka的動態擴容是經過Zookeeper來實現的。

Zookeeper是一種在分佈式系統中被普遍用來做爲：分佈式狀態管理、分佈式協調管理、分佈式配置管理、和分佈式鎖服務的集羣。kafka增長和減小服務器都會在Zookeeper節點上觸發相應的事件kafka系統會捕獲這些事件，進行新一輪的負載均衡，客戶端也會捕獲這些事件來進行新一輪的處理。

Kafka相關概念

一、 AMQP協議java

Advanced Message Queuing Protocol （高級消息隊列協議）

The Advanced Message Queuing Protocol (AMQP)： 是一個標準開放的應用層的消息中間件（Message Oriented Middleware）協議。AMQP定義了經過網絡發送的字節流的數據格式。所以兼容性很是好，任何實現AMQP協議的程序均可以和與AMQP協議兼容的其餘程序交互，能夠很容易作到跨語言，跨平臺。

上面說的3種比較流行的消息隊列協議，要麼支持AMQP協議，要麼借鑑了AMQP協議的思想進行了開發、實現、設計。

二、一些基本的概念

一、消費者：（Consumer）：從消息隊列中請求消息的客戶端應用程序

二、生產者：（Producer）：向broker發佈消息的應用程序

三、AMQP服務端（broker）：用來接收生產者發送的消息並將這些消息路由給服務器中的隊列，便於fafka將生產者發送的消息，動態的添加到磁盤並給每一條消息一個偏移量，因此對於kafka一個broker就是一個應用程序的實例

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

可使用以上任何一種語言和kafka服務器進行通訊（即辨析本身的consumer從kafka集羣訂閱消息也能夠本身寫producer程序）

三、Kafka架構

生產者生產消息、kafka集羣、消費者獲取消息這樣一種架構，以下圖：

kafka集羣中的消息，是經過Topic（主題）來進行組織的，以下圖：

一些基本的概念：

一、主題（Topic）：一個主題相似新聞中的體育、娛樂、教育等分類概念，在實際工程中一般一個業務一個主題。

二、分區（Partition）：一個Topic中的消息數據按照多個分區組織，分區是kafka消息隊列組織的最小單位，一個分區能夠看做是一個FIFO（ First Input First Output的縮寫，先入先出隊列）的隊列。

kafka分區是提升kafka性能的關鍵所在，當你發現你的集羣性能不高時，經常使用手段就是增長Topic的分區，分區裏面的消息是按照重新到老的順序進行組織，消費者從隊列頭訂閱消息，生產者從隊列尾添加消息。

工做圖：

備份（Replication）：爲了保證分佈式可靠性，kafka0.8開始對每一個分區的數據進行備份（不一樣的Broker上），防止其中一個Broker宕機形成分區上的數據不可用。

kafka0.7是一個很大的改變：一、增長了備份二、增長了控制借點概念，增長了集羣領導者選舉。

Zookeeper集羣搭建

Kafka集羣是把狀態保存在Zookeeper中的，首先要搭建Zookeeper集羣。

一、軟件環境

（3臺服務器-個人測試）

192.168.7.100 server1

192.168.7.101 server2

192.168.7.107 server3

一、Linux服務器一臺、三臺、五臺、（2*n+1），Zookeeper集羣的工做是超過半數才能對外提供服務，3臺中超過兩臺超過半數，容許1臺掛掉，是否能夠用偶數，其實不必。

若是有四臺那麼掛掉一臺還剩下三臺服務器，若是在掛掉一個就不行了，這裏記住是超過半數。

二、Java jdk1.7 zookeeper是用java寫的因此他的須要JAVA環境，java是運行在java虛擬機上的

三、Zookeeper的穩定版本Zookeeper 3.4.6版本

二、配置&安裝Zookeeper

下面的操做是：3臺服務器統一操做

一、安裝Java

yum list java*
yum -y install java-1.7.0-openjdk*

二、下載Zookeeperlinux

首先要注意在生產環境中目錄結構要定義好，防止在項目過多的時候找不到所需的項目數據庫

#個人目錄統一放在/opt下面
#首先建立Zookeeper項目目錄
mkdir zookeeper #項目目錄
mkdir zkdata #存放快照日誌
mkdir zkdatalog#存放事物日誌

下載Zookeeperapache

#下載軟件
cd /opt/zookeeper/

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

#解壓軟件
tar -zxvf zookeeper-3.4.6.tar.gz

三、修改配置文件bash

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

#進入conf目錄
/opt/zookeeper/zookeeper-3.4.6/conf
#查看
[root@192.168.7.107]$ ll
-rw-rw-r--. 1 1000 1000  535 Feb 20  2014 configuration.xsl
-rw-rw-r--. 1 1000 1000 2161 Feb 20  2014 log4j.properties
-rw-rw-r--. 1 1000 1000  922 Feb 20  2014 zoo_sample.cfg

#zoo_sample.cfg 這個文件是官方給咱們的zookeeper的樣板文件，給他複製一份命名爲zoo.cfg，zoo.cfg是官方指定的文件命名規則。服務器

3臺服務器的配置文件網絡

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkdata
dataLogDir=/opt/zookeeper/zkdatalog
clientPort=12181
server.1=192.168.7.100:12888:13888
server.2=192.168.7.101:12888:13888
server.3=192.168.7.107:12888:13888
#server.1 這個1是服務器的標識也能夠是其餘的數字， 表示這個是第幾號服務器，用來標識服務器，這個標識要寫到快照目錄下面myid文件裏
#192.168.7.107爲集羣裏的IP地址，第一個端口是master和slave之間的通訊端口，默認是2888，第二個端口是leader選舉的端口，集羣剛啓動的時候選舉或者leader掛掉以後進行新的選舉的端口默認是3888

配置文件解釋：架構

#tickTime：
這個時間是做爲 Zookeeper 服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每一個 tickTime 時間就會發送一個心跳。
#initLimit：
這個配置項是用來配置 Zookeeper 接受客戶端（這裏所說的客戶端不是用戶鏈接 Zookeeper 服務器的客戶端，而是 Zookeeper 服務器集羣中鏈接到 Leader 的 Follower 服務器）初始化鏈接時最長能忍受多少個心跳時間間隔數。當已經超過 5個心跳的時間（也就是 tickTime）長度後 Zookeeper 服務器尚未收到客戶端的返回信息，那麼代表這個客戶端鏈接失敗。總的時間長度就是 5*2000=10 秒
#syncLimit：
這個配置項標識 Leader 與Follower 之間發送消息，請求和應答時間長度，最長不能超過多少個 tickTime 的時間長度，總的時間長度就是5*2000=10秒
#dataDir：
快照日誌的存儲路徑
#dataLogDir：
事物日誌的存儲路徑，若是不配置這個那麼事物日誌會默認存儲到dataDir制定的目錄，這樣會嚴重影響zk的性能，當zk吞吐量較大的時候，產生的事物日誌、快照日誌太多
#clientPort：
這個端口就是客戶端鏈接 Zookeeper 服務器的端口，Zookeeper 會監聽這個端口，接受客戶端的訪問請求。修改他的端口改大點

建立myid文件併發

#server1
echo "1" > /opt/zookeeper/zkdata/myid
#server2
echo "2" > /opt/zookeeper/zkdata/myid
#server3
echo "3" > /opt/zookeeper/zkdata/myid

四、重要配置說明

一、myid文件和server.myid 在快照目錄下存放的標識本臺服務器的文件，他是整個zk集羣用來發現彼此的一個重要標識。

二、zoo.cfg 文件是zookeeper配置文件在conf目錄裏。

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

# Define some default values that can be overridden by system properties
zookeeper.root.logger=INFO, CONSOLE  #日誌級別
zookeeper.console.threshold=INFO  #使用下面的console來打印日誌
zookeeper.log.dir=.    #日誌打印到那裏，是我們啓動zookeeper的目錄 （建議設置統一的日誌目錄路徑）
zookeeper.log.file=zookeeper.log
zookeeper.log.threshold=DEBUG
zookeeper.tracelog.dir=.
zookeeper.tracelog.file=zookeeper_trace.log

#
# ZooKeeper Logging Configuration
#

# Format is "<default threshold> (, <appender>)+

# DEFAULT: console appender only
log4j.rootLogger=${zookeeper.root.logger}

# Example with rolling log file
#log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE

# Example with rolling log file and tracing
#log4j.rootLogger=TRACE, CONSOLE, ROLLINGFILE, TRACEFILE

#
# Log INFO level and above messages to the console
#
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Threshold=${zookeeper.console.threshold}
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n


# Add ROLLINGFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold}
log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file}

# Max log file size of 10MB
log4j.appender.ROLLINGFILE.MaxFileSize=10MB
# uncomment the next line to limit number of backup files
#log4j.appender.ROLLINGFILE.MaxBackupIndex=10

log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n


#
# Add TRACEFILE to rootLogger to get log file output
#    Log DEBUG level and above messages to a log file
log4j.appender.TRACEFILE=org.apache.log4j.FileAppender
log4j.appender.TRACEFILE.Threshold=TRACE
log4j.appender.TRACEFILE.File=${zookeeper.tracelog.dir}/${zookeeper.tracelog.file}

log4j.appender.TRACEFILE.layout=org.apache.log4j.PatternLayout
### Notice we are including log4j's NDC here (%x)
log4j.appender.TRACEFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L][%x] - %m%n

四、zkEnv.sh和zkServer.sh文件

zkServer.sh 主的管理程序文件

zkEnv.sh 是主要配置，zookeeper集羣啓動時配置環境變量的文件

五、還有一個須要注意

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不會主動的清除舊的快照和日誌文件，這個是操做者的責任。

可是能夠經過命令去按期的清理。

#!/bin/bash 
 
#snapshot file dir 
dataDir=/opt/zookeeper/zkdata/version-2
#tran log dir 
dataLogDir=/opt/zookeeper/zkdatalog/version-2

#Leave 66 files 
count=66 
count=$[$count+1] 
ls -t $dataLogDir/log.* | tail -n +$count | xargs rm -f 
ls -t $dataDir/snapshot.* | tail -n +$count | xargs rm -f 

#以上這個腳本定義了刪除對應兩個目錄中的文件，保留最新的66個文件，能夠將他寫到crontab中，設置爲天天凌晨2點執行一次就能夠了。


#zk log dir   del the zookeeper log
#logDir=
#ls -t $logDir/zookeeper.log.* | tail -n +$count | xargs rm -f

其餘方法：

第二種：使用ZK的工具類PurgeTxnLog，它的實現了一種簡單的歷史文件清理策略，能夠在這裏看一下他的使用方法 http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html

第三種：對於上面這個執行，ZK本身已經寫好了腳本，在bin/zkCleanup.sh中，因此直接使用這個腳本也是能夠執行清理工做的。

第四種：從3.4.0開始，zookeeper提供了自動清理snapshot和事務日誌的功能，經過配置 autopurge.snapRetainCount 和 autopurge.purgeInterval 這兩個參數可以實現定時清理了。這兩個參數都是在zoo.cfg中配置的：

autopurge.purgeInterval 這個參數指定了清理頻率，單位是小時，須要填寫一個1或更大的整數，默認是0，表示不開啓本身清理功能。

autopurge.snapRetainCount 這個參數和上面的參數搭配使用，這個參數指定了須要保留的文件數目。默認是保留3個。

推薦使用第一種方法，對於運維人員來講，將日誌清理工做獨立出來，便於統一管理也更可控。畢竟zk自帶的一些工具並不怎麼給力。

五、啓動服務並查看

一、啓動服務

#進入到Zookeeper的bin目錄下
cd /opt/zookeeper/zookeeper-3.4.6/bin
#啓動服務（3臺都須要操做）
./zkServer.sh start

二、檢查服務狀態

#檢查服務器狀態
./zkServer.sh status

經過status就能看到狀態：

./zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg  #配置文件
Mode: follower  #他是否爲領導

zk集羣通常只有一個leader，多個follower，主通常是相應客戶端的讀寫請求，而從主同步數據，當主掛掉以後就會從follower裏投票選舉一個leader出來。

能夠用「jps」查看zk的進程，這個是zk的整個工程的main

#執行命令jps
20348 Jps
4233 QuorumPeerMain