環境：
CentOS 6.5
KAFKA版本： kafka_2.11-0.8.2.1
ZOOKEEPER版本： zookeeper-3.4.6
JDK： 1.8.0_151
SERVER: 172.16.2.27、172.16.2.28、172.16.2.29
一、準備：
1、安裝JDK1.8
2、下載下傳kafka和zookeeper安裝包（二進制包）
下載下傳位址：
二、安裝zookeeper叢集
1
2
3
4
5
6
7
8
9
10
11
12
<code>##将安裝檔案解壓縮到/usr/local目錄</code>
<code>tar</code> <code>zxf zookeeper-3.4.6.</code><code>tar</code><code>.gz -C </code><code>/usr/local</code>
<code>##建立軟連接配接</code>
<code>ln</code> <code>-s </code><code>/usr/local/zookeeper-3</code><code>.4.6 </code><code>/usr/local/zookeeper</code>
<code>##從模闆複制配置檔案</code>
<code>cd</code> <code>/usr/local/zookeeper/conf</code>
<code>cp</code> <code>zoo_sample.cfg zoo.cfg</code>
<code>##建立data目錄</code>
<code>mkdir</code> <code>-p </code><code>/data/zookeeper/</code><code>{zkdata,zkdatalog}</code>
修改配置檔案
13
14
15
16
<code>cat</code> <code>zoo.cfg | </code><code>egrep</code> <code>-</code><code>v</code> <code>"^#|^$"</code>
<code>tickTime=2000 </code>
<code>initLimit=10</code>
<code>syncLimit=5</code>
<code>dataDir=</code><code>/data/zookeeper/zkdata</code>
<code>dataLogDir=</code><code>/data/zookeeper/zkdatalog</code>
<code>clientPort=2181</code>
<code>server.1=172.16.2.27:2888:3888</code>
<code>server.2=172.16.2.28:2888:3888</code>
<code>server.3=172.16.2.29:2888:3888</code>
<code>autopurge.snapRetainCount=30</code>
<code>autopurge.purgeInterval=24</code>
<code>###server.1 這個1是伺服器的辨別也可以是其他的數字， 表示這個是第幾号伺服器，用來辨別伺服器</code>
<code>這個辨別要寫到快照目錄下面myid檔案裡</code>
<code>#172.16.2.27為叢集裡的IP位址，第一個端口是master和slave之間的通信端口，預設是2888，</code>
<code>第二個端口是leader選舉的端口，叢集剛啟動的時候選舉或者leader挂掉之後進行新的選舉的端口預設是3888</code>
三台伺服器上的配置是一樣的
配置檔案解釋：
<code>#tickTime：</code>
<code>這個時間是作為 Zookeeper 伺服器之間或用戶端與伺服器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。</code>
<code>#initLimit：</code>
<code>這個配置項是用來配置 Zookeeper 接受用戶端（這裡所說的用戶端不是使用者連接配接 Zookeeper 伺服器的用戶端，而是 Zookeeper 伺服器叢集中連接配接到 Leader 的 Follower 伺服器）初始化連接配接時最長能忍受多少個心跳時間間隔數。</code>
<code>當已經超過 5個心跳的時間（也就是 tickTime）長度後 Zookeeper 伺服器還沒有收到用戶端的傳回資訊，那麼表明這個用戶端連接配接失敗。總的時間長度就是 5*2000=10 秒</code>
<code>#syncLimit：</code>
<code>這個配置項辨別 Leader 與Follower 之間發送消息，請求和應答時間長度，最長不能超過多少個 tickTime 的時間長度，總的時間長度就是5*2000=10秒</code>
<code>#dataDir：</code>
<code>快照日志的存儲路徑</code>
<code>#dataLogDir：</code>
<code>事物日志的存儲路徑，如果不配置這個那麼事物日志會預設存儲到dataDir制定的目錄，這樣會嚴重影響zk的性能，當zk吞吐量較大的時候，産生的事物日志、快照日志太多</code>
<code>#clientPort：</code>
<code>這個端口就是用戶端連接配接 Zookeeper 伺服器的端口，Zookeeper 會監聽這個端口，接受用戶端的通路請求。修改他的端口改大點</code>
建立myid檔案
<code>#server1</code>
<code>echo</code> <code>"1"</code> <code>> </code><code>/data/zookeeper/zkdata/myid</code>
<code>#server2</code>
<code>echo</code> <code>"2"</code> <code>> </code><code>/data/zookeeper/zkdata/myid</code>
<code>#server3</code>
<code>echo</code> <code>"3"</code> <code>></code><code>/data/zookeeper/zkdata/myid</code>
重要說明：
1、myid檔案和server.myid 在快照目錄下存放的辨別本台伺服器的檔案，他是整個zk叢集用來發現彼此的一個重要辨別。
2、zoo.cfg 檔案是zookeeper配置檔案 在conf目錄裡。
3、log4j.properties檔案是zk的日志輸出檔案 在conf目錄裡用java寫的程式基本上有個共同點日志都用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不會主動的清除舊的快照和日志檔案，這個是操作者的責任。
啟動ZooKeeper服務
<code>cd</code> <code>/usr/local/zookeeper/bin</code>
<code>.</code><code>/zkServer</code><code>.sh start</code>
<code>##檢視狀态</code>
<code>.</code><code>/zkServer</code><code>.sh status</code>
<code>JMX enabled by default</code>
<code>Using config: </code><code>/usr/local/zookeeper/bin/</code><code>..</code><code>/conf/zoo</code><code>.cfg </code><code>##zookeeper使用的配置檔案</code>
<code>Mode: follower </code><code>##zookeeper的角色</code>
三、安裝KAFKA叢集
<code>tar</code> <code>zxf kafka_2.11-0.8.2.1.tgz -C </code><code>/usr/local</code>
<code>ln</code> <code>-s </code><code>/usr/local/kafka_2</code><code>.11-0.8.2.1 </code><code>/usr/local/kafka</code>
<code>cd</code> <code>/usr/local/kafka</code>
<code>cp</code> <code>server.properties server.properties.bak</code>
17
18
<code>broker.</code><code>id</code><code>=0 </code><code>#目前機器在叢集中的唯一辨別，和zookeeper的myid性質一樣</code>
<code>port=19092 </code><code>#目前kafka對外提供服務的端口預設是9092</code>
<code>host.name=172.16.2.27 </code><code>#這個參數預設是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。</code>
<code>num.network.threads=3 </code><code>#這個是borker進行網絡處理的線程數</code>
<code>num.io.threads=8 </code><code>#這個是borker進行I/O處理的線程數</code>
<code>log.</code><code>dirs</code><code>=log.</code><code>dirs</code><code>=</code><code>/data/kafka/kafka-logs</code> <code>#消息存放的目錄，這個目錄可以配置為“，”逗号分割的表達式，上面的num.io.threads要大于這個目錄的個數這個目錄，如果配置多個目錄，新建立的topic他把消息持久化的地方是，目前以逗号分割的目錄中，那個分區數最少就放那一個</code>
<code>socket.send.buffer.bytes=102400 </code><code>#發送緩沖區buffer大小，資料不是一下子就發送的，先回存儲到緩沖區了到達一定的大小後在發送，能提高性能</code>
<code>socket.receive.buffer.bytes=102400 </code><code>#kafka接收緩沖區大小，當資料到達一定大小後在序列化到磁盤</code>
<code>socket.request.max.bytes=104857600 </code><code>#這個參數是向kafka請求消息或者向kafka發送消息的請請求的最大數，這個值不能超過java的堆棧大小</code>
<code>num.partitions=1 </code><code>#預設的分區數，一個topic預設1個分區數</code>
<code>log.retention.hours=168 </code><code>#預設消息的最大持久化時間，168小時，7天</code>
<code>message.max.byte=5242880 </code><code>#消息儲存的最大值5M</code>
<code>default.replication.factor=2 </code><code>#kafka儲存消息的副本數，如果一個副本失效了，另一個還可以繼續提供服務</code>
<code>replica.fetch.max.bytes=5242880 </code><code>#取消息的最大直接數</code>
<code>log.segment.bytes=1073741824 </code><code>#這個參數是：因為kafka的消息是以追加的形式落地到檔案，當超過這個值的時候，kafka會新起一個檔案</code>
<code>log.retention.check.interval.ms=300000 </code><code>#每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄檢視是否有過期的消息如果有，删除</code>
<code>log.cleaner.</code><code>enable</code><code>=</code><code>false</code> <code>#是否啟用log壓縮，一般不用啟用，啟用的話可以提高性能</code>
<code>zookeeper.connect=172.16.2.27:2181,172.16.2.28:2181,172.16.2.29:2181 </code><code>#設定zookeeper的連接配接端口</code>
<code>broker.</code><code>id</code><code>=0 每台伺服器的broker.</code><code>id</code><code>都不能相同</code>
<code>host.name=172.16.2.27</code>
<code>#在log.retention.hours=168 下面新增下面三項</code>
<code>message.max.byte=5242880</code>
<code>default.replication.factor=2</code>
<code>replica.fetch.max.bytes=5242880</code>
<code>#設定zookeeper的連接配接端口</code>
<code>zookeeper.connect=172.16.2.27:2181,172.16.2.28:2181,172.16.2.29:2181</code>
啟動kafka服務并測試：
<code>#從背景啟動Kafka叢集（3台都需要啟動）</code>
<code>cd</code> <code>/usr/local/kafka/bin/</code> <code>#進入到kafka的bin目錄 </code>
<code>.</code><code>/kafka-server-start</code><code>.sh -daemon ..</code><code>/config/server</code><code>.properties</code>
檢查kafka服務是否啟動
<code># jps</code>
<code>67137 Kafka</code>
<code>111089 ProdServerStart</code>
<code>122216 Jps</code>
<code>114635 QuorumPeerMain</code>
三台伺服器上配置基本一樣，除了下面兩條配置不同
<code>broker.</code><code>id</code><code>=0</code>
<code></code>
本文轉自 曾哥最愛 51CTO部落格，原文連結：http://blog.51cto.com/zengestudy/2049745，如需轉載請自行聯系原作者
![Java小案例——随機數猜測随機數猜測[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)