#進入conf目錄
/opt/zookeeper/zookeeper-3.4.6/conf
#檢視
[[email protected]]$ 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

 4、重要配置說明

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

2、zoo.cfg 檔案是zookeeper配置檔案 在conf目錄裡。

3、log4j.properties檔案是zk的日誌輸出檔案 在conf目錄裡用java寫的程式基本上有個共同點日誌都用log4j，來進行管理。

 configuration for log4j

4、zkEnv.sh和zkServer.sh檔案

zkServer.sh 主的管理程式檔案 zkEnv.sh 是主要配置，zookeeper叢集啟動時配置環境變數的檔案 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不會主動的清除舊的快照和日誌檔案，這個是操作者的責任。

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

#!/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自帶的一些工具並不怎麼給力。 5、啟動服務並檢視 1、啟動服務

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

2、檢查服務狀態

#檢查伺服器狀態
./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 

Kafka叢集搭建

1、軟體環境 1、linux一臺或多臺，大於等於2 2、已經搭建好的zookeeper叢集 3、軟體版本kafka_2.11-0.9.0.1.tgz 2、建立目錄並下載安裝軟體

#建立目錄
cd /opt/
mkdir kafka #建立專案目錄
cd kafka
mkdir kafkalogs #建立kafka訊息目錄，主要存放kafka訊息

#下載軟體
wget  http://apache.opencas.org/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz

#解壓軟體
tar -zxvf kafka_2.11-0.9.0.1.tgz

3、修改配置檔案

進入到config目錄

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

主要關注：server.properties 這個檔案即可，我們可以發現在目錄下：

有很多檔案，這裡可以發現有Zookeeper檔案，我們可以根據Kafka內帶的zk叢集來啟動，但是建議使用獨立的zk叢集

-rw-r--r--. 1 root root 5699 Feb 22 09:41 192.168.7.101
-rw-r--r--. 1 root root  906 Feb 12 08:37 connect-console-sink.properties
-rw-r--r--. 1 root root  909 Feb 12 08:37 connect-console-source.properties
-rw-r--r--. 1 root root 2110 Feb 12 08:37 connect-distributed.properties
-rw-r--r--. 1 root root  922 Feb 12 08:38 connect-file-sink.properties
-rw-r--r--. 1 root root  920 Feb 12 08:38 connect-file-source.properties
-rw-r--r--. 1 root root 1074 Feb 12 08:37 connect-log4j.properties
-rw-r--r--. 1 root root 2055 Feb 12 08:37 connect-standalone.properties
-rw-r--r--. 1 root root 1199 Feb 12 08:37 consumer.properties
-rw-r--r--. 1 root root 4369 Feb 12 08:37 log4j.properties
-rw-r--r--. 1 root root 2228 Feb 12 08:38 producer.properties
-rw-r--r--. 1 root root 5699 Feb 15 18:10 server.properties
-rw-r--r--. 1 root root 3325 Feb 12 08:37 test-log4j.properties
-rw-r--r--. 1 root root 1032 Feb 12 08:37 tools-log4j.properties
-rw-r--r--. 1 root root 1023 Feb 12 08:37 zookeeper.properties

修改配置檔案：

broker.id=0  #當前機器在叢集中的唯一標識，和zookeeper的myid性質一樣
port=19092 #當前kafka對外提供服務的埠預設是9092
host.name=192.168.7.100 #這個引數預設是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。
num.network.threads=3 #這個是borker進行網路處理的執行緒數
num.io.threads=8 #這個是borker進行I/O處理的執行緒數
log.dirs=/opt/kafka/kafkalogs/ #訊息存放的目錄，這個目錄可以配置為“，”逗號分割的表示式，上面的num.io.threads要大於這個目錄的個數這個目錄，如果配置多個目錄，新建立的topic他把訊息持久化的地方是，當前以逗號分割的目錄中，那個分割槽數最少就放那一個
socket.send.buffer.bytes=102400 #傳送緩衝區buffer大小，資料不是一下子就傳送的，先回儲存到緩衝區了到達一定的大小後在傳送，能提高效能
socket.receive.buffer.bytes=102400 #kafka接收緩衝區大小，當資料到達一定大小後在序列化到磁碟
socket.request.max.bytes=104857600 #這個引數是向kafka請求訊息或者向kafka傳送訊息的請請求的最大數，這個值不能超過java的堆疊大小
num.partitions=1 #預設的分割槽數，一個topic預設1個分割槽數
log.retention.hours=168 #預設訊息的最大持久化時間，168小時，7天
message.max.byte=5242880  #訊息儲存的最大值5M
default.replication.factor=2  #kafka儲存訊息的副本數，如果一個副本失效了，另一個還可以繼續提供服務
replica.fetch.max.bytes=5242880  #取訊息的最大直接數
log.segment.bytes=1073741824 #這個引數是：因為kafka的訊息是以追加的形式落地到檔案，當超過這個值的時候，kafka會新起一個檔案
log.retention.check.interval.ms=300000 #每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄檢視是否有過期的訊息如果有，刪除
log.cleaner.enable=false #是否啟用log壓縮，一般不用啟用，啟用的話可以提高效能
zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:1218 #設定zookeeper的連線埠

上面是引數的解釋，實際的修改項為：

#broker.id=0  每臺伺服器的broker.id都不能相同


#hostname
host.name=192.168.7.100

#在log.retention.hours=168 下面新增下面三項
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880

#設定zookeeper的連線埠
zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:12181

4、啟動Kafka叢集並測試

1、啟動服務

#從後臺啟動Kafka叢集（3臺都需要啟動）
cd

/opt/kafka/kafka_2.11-0.9.0.1//bin #進入到kafka的bin目錄 
./kafka-server-start.sh -daemon ../config/server.properties

2、檢查服務是否啟動

#執行命令jps
20348 Jps
4233 QuorumPeerMain
18991 Kafka

3、建立Topic來驗證是否建立成功

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

#建立Topic
./kafka-topics.sh --create --zookeeper 192.168.7.100:12181 --replication-factor 2 --partitions 1 --topic shuaige
#解釋
--replication-factor 2   #複製兩份
--partitions 1 #建立1個分割槽
--topic #主題為shuaige

'''在一臺伺服器上建立一個釋出者'''
#建立一個broker，釋出者
./kafka-console-producer.sh --broker-list 192.168.7.100:19092 --topic shuaige

'''在一臺伺服器上建立一個訂閱者'''
./kafka-console-consumer.sh --zookeeper localhost:12181 --topic shuaige --from-beginning

測試（在釋出者那裡釋出訊息看看訂閱者那裡是否能正常收到~）：

4、其他命令

大部分命令可以去官方文件檢視

4.1、檢視topic

./kafka-topics.sh --list --zookeeper localhost:12181
#就會顯示我們建立的所有topic

4.2、檢視topic狀態

/kafka-topics.sh --describe --zookeeper localhost:12181 --topic shuaige
#下面是顯示資訊
Topic:ssports    PartitionCount:1    ReplicationFactor:2    Configs:
    Topic: shuaige    Partition: 0    Leader: 1    Replicas: 0,1    Isr: 1
#分割槽為為1  複製因子為2   他的  shuaige的分割槽為0 
#Replicas: 0,1   複製的為0，1
# 

 OKkafka叢集搭建完畢

5、其他說明標註

5.1、日誌說明

預設kafka的日誌是儲存在/opt/kafka/kafka_2.10-0.9.0.0/logs目錄下的，這裡說幾個需要注意的日誌

server.log #kafka的執行日誌
state-change.log  #kafka他是用zookeeper來儲存狀態，所以他可能會進行切換，切換的日誌就儲存在這裡

controller.log #kafka選擇一個節點作為“controller”,當發現有節點down掉的時候它負責在游泳分割槽的所有節點中選擇新的leader,這使得Kafka可以批量的高效的管理所有分割槽節點的主從關係。如果controller down掉了，活著的節點中的一個會備切換為新的controller.

5.2、上面的大家你完成之後可以登入zk來檢視zk的目錄情況

#使用客戶端進入zk
./zkCli.sh -server 127.0.0.1:12181  #預設是不用加’-server‘引數的因為我們修改了他的埠

#檢視目錄情況 執行“ls /”
[zk: 127.0.0.1:12181(CONNECTED) 0] ls /

#顯示結果：[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch]
'''
上面的顯示結果中：只有zookeeper是，zookeeper原生的，其他都是Kafka建立的
'''

#標註一個重要的
[zk: 127.0.0.1:12181(CONNECTED) 1] get /brokers/ids/0
{"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]