Kafka集群搭建

本文轉載自查看原文 2017-06-26 09:08 3675

Kafka初識

Kafka使用背景

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

我們想分析下用戶行為（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協議

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是一個很大的改變：1、增加了備份2、增加了控制借點概念，增加了集群領導者選舉。

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*
下載Zookeeper

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

#我的目錄統一放在/opt下面
#首先創建Zookeeper項目目錄
mkdir zookeeper #項目目錄
mkdir zkdata #存放快照日志
mkdir zkdatalog#存放事物日志

下載Zookeeper

#下載軟件
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

修改配置文件

進入到解壓好的目錄里面的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，來進行管理。
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