為了演示集群的效果，這里准備一台虛擬機(window 7)，在虛擬機中搭建了單IP多節點的zookeeper集群（多IP節點的也是同理的），並且在本機（win 7）和虛擬機中都安裝了kafka。
前期准備說明：
1.三台zookeeper服務器，本機安裝一個作為server1，虛擬機安裝兩個(單IP)
2.三台kafka服務器，本機安裝一個作為server1，虛擬機安裝兩個。
備注：當然你可以直接在虛擬機上安裝三個服務器分別為server1、server2、server3 。
- 虛擬機和本機網絡環境
1.將虛擬機的網絡模式調整為橋接模式，將虛擬機的防火牆功能關閉；
2.將主機的防火牆也關閉。
3.互相ping，測試是否能正常連接.。
- 下載kafka
從Kafka官網http://kafka.apache.org/downloads下載Kafka安裝包。(要下載Binary downloads這個類型，不要下載源文件，方便使用)
- 下載zookeeper
從zookeeper官網http://zookeeper.apache.org/releases.html下載zookeeper安裝包。
- 下載JDK
從官網http://www.java.com/download/下載(建議下載Oracle官方發布的Java),，配置環境變量。
zookeeper集群環境搭建
本機配置
修改本機sever1的zoo.cfg文件配置:
#存儲內存中數據庫快照的位置，如果不設置參數，更新事務日志將被存儲到默認位置。 #dataDir=/tmp/zookeeper dataDir=D:/bigData/zookeeper-3.4.10/data #日志文件的位置 dataLogDir=D:/bigData/zookeeper-3.4.10/zlog #監聽端口 clientPort=2181 #集群服務器配置 server.1=192.168.1.130:9000:7000 server.2=192.168.1.103:9001:7001 server.3=192.168.1.103:9002:7002
格式： server.A = B:C:D
A:是一個數字，表示第幾號服務器
B:服務器IP地址
C:是一個端口號，用來集群成員的信息交換,表示這個服務器與集群中的leader服務器交換信息的端口
D:是在leader掛掉時專門用來進行選舉leader所用的端口
完整配置文件如下：
# The number of milliseconds of each tick tickTime=2000 # The number of ticks that the initial # synchronization phase can take initLimit=10 # The number of ticks that can pass between # sending a request and getting an acknowledgement syncLimit=5 # the directory where the snapshot is stored. # do not use /tmp for storage, /tmp here is just # example sakes. #dataDir=/tmp/zookeeper dataDir=D:/bigData/zookeeper-3.4.10/data dataLogDir=D:/bigData/zookeeper-3.4.10/zlog # the port at which the clients will connect clientPort=2181 # the maximum number of client connections. # increase this if you need to handle more clients #maxClientCnxns=60 # # Be sure to read the maintenance section of the # administrator guide before turning on autopurge. # # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance # # The number of snapshots to retain in dataDir #autopurge.snapRetainCount=3 # Purge task interval in hours # Set to "0" to disable auto purge feature #autopurge.purgeInterval=1 server.1=192.168.1.130:9000:7000 server.2=192.168.1.101:9001:7001 server.3=192.168.1.101:9002:7002
創建serverID
在zoo.cfg配置文件中dataDir目錄中新建一個沒有后綴的myid文件，里邊寫1。
虛擬機配置
將本機的zookeeper安裝包拷貝到虛擬機上，然后在盤中新建兩個目錄，server2,server3目錄。
修改server2的zoo.cfg配置文件：
#存儲內存中數據庫快照的位置，如果不設置參數，更新事務日志將被存儲到默認位置。 #dataDir=/tmp/zookeeper dataDir=E:/bigData/server2/zookeeper-3.4.10/data #日志文件的位置 dataLogDir=E:/bigData/server2/zookeeper-3.4.10/zlog #監聽端口 clientPort=2182 #集群服務器配置 server.1=192.168.1.130:9000:7000 server.2=192.168.1.103:9001:7001 server.3=192.168.1.103:9002:7002
創建SeverID
在zoo.cfg配置文件中dataDir目錄中新建一個沒有后綴的myid文件，里邊寫2。
同理配置server3中的zookeeper，這樣zookeeper集群的配置工作已經完成。
下面我們啟動zookeeper集群：一個一個啟動每一個zookeeper服務器：
先后進入每一個zookeeper服務器的bin目錄下執行zkserver.cmd,在先啟動的zookeeper會打印其他zookeeper服務器未啟動的錯誤信息，直至到最后一個啟動的時候就沒有異常信息了。正常界面：
zookeeper集群環境搭建遇到問題總結:
- zookeeper啟動時報Cannot open channel to X at election address Error contacting service. It is probably not running.
解決方法：
a. 關閉所有服務器防火牆
b. 將本服務器的IP修改成0.0.0.0
例如：server.1=0.0.0.0:9000:7000
Kafka集群環境搭建：
下載kafka后解壓到某個目錄，然后修改server.properties配置。
server.properties修改說明：
broker.id=0 • 在kafka這個集群中的唯一標識，且只能是正整數 port=9091 • 該服務監聽的端口 host.name=192.168.1.130 • broker 綁定的主機名稱(IP) 如果不設置將綁定所有的接口。 advertised.host.name=192.168.1.130 • broker服務將通知消費者和生產者 換言之，就是消費者和生產者就是通過這個主機(IP)來進行通信的。如果沒有設置就默認采用host.name。 num.network.threads=2 • broker處理消息的最大線程數，一般情況是CPU的核數 num.io.threads=8 • broker處理IO的線程數 一般是num.network.threads的兩倍 socket.send.buffer.bytes=1048576 • socket發送的緩沖區。socket調優參數SO_SNDBUFF socket.receive.buffer.bytes=1048576 • socket接收的緩沖區 socket的調優參數SO_RCVBUF socket.request.max.bytes=104857600 • socket請求的最大數量，防止serverOOM。 log.dirs=\logs • kafka數據的存放地址，多個地址的話用逗號隔開。多個目錄分布在不同的磁盤上可以提高讀寫性能 num.partitions=2 • 每個tipic的默認分區個數，在創建topic時可以重新制定 log.retention.hours=168 • 數據文件的保留時間 log.retention.minutes也是一個道理。 log.segment.bytes=536870912 • topic中的最大文件的大小 -1表示沒有文件大小限制 log.segment.bytes 和log.retention.minutes 任意一個 達到要求 都會刪除該文件 在創建topic時可以重新制定。若沒有.則選取該默認值 log.retention.check.interval.ms=60000 • 文件大小檢查的周期時間，是否處罰 log.cleanup.policy中設置的策略 log.cleaner.enable=false • 是否開啟日志清理 zookeeper.connect=192.168.1.130:num1,192.168.1.130:num2,192.168.1.130:num3 • 上面我們的Zookeeper集群 zookeeper.connection.timeout.ms=1000000 • 進群鏈接時間超時
本機配置
修改本機server1的server.properties配置：
#在kafka這個集群中的唯一標識，且只能是正整數 broker.id=0 # kafka集群的地址 broker.list=192.168.0.130:9092,192.168.0.103:9093,192.168.0.103:9094 #listener and port port=9092 #broker 綁定的主機名稱(IP) 如果不設置將綁定所有的接口。 host.name=192.168.1.130 #kafka數據的存放地址，多個地址的話用逗號隔開。多個目錄分布在不同的磁盤上可以提高讀寫性能 # A comma separated list of directories under which to store log files #log.dirs=/tmp/kafka-logs log.dirs=D:/bigData/kafka_2.11-1.1.0/kafka-logs #每個tipic的默認分區個數，在創建topic時可以重新制定 num.partitions=3 # root directory for all kafka znodes. zookeeper.connect=192.168.1.130:2181,192.168.1.103:2182,192.168.1.103:2183
虛擬機配置
將本機的kafka安裝包拷貝到虛擬機上，存放在server2,server3目錄。
注意：若本機拷貝到虛擬機中kafka之前有使用過，生成過topics，直接拷貝搭建集群，啟動kafka時會報如下錯誤：
原因：在單點環境下創建的topic中，kafka服務器的broker.id為0，在搭建集群時修改了borker.id，先前創建的topic就無法找到對應的broker.id。
解決辦法：刪除原來的topic信息。
修改server2目錄中kafka配置：
#在kafka這個集群中的唯一標識，且只能是正整數 broker.id=1 # kafka集群的地址 broker.list=192.168.0.130:9092,192.168.0.103:9093,192.168.0.103:9094 #listener and port port=9093 #broker 綁定的主機名稱(IP) 如果不設置將綁定所有的接口。 host.name=192.168.1.103 #kafka數據的存放地址，多個地址的話用逗號隔開。多個目錄分布在不同的磁盤上可以提高讀寫性能 # A comma separated list of directories under which to store log files #log.dirs=/tmp/kafka-logs log.dirs=D:/bigData/server2/kafka_2.11-1.1.0/kafka-logs #每個tipic的默認分區個數，在創建topic時可以重新制定 num.partitions=3 # root directory for all kafka znodes. zookeeper.connect=192.168.1.130:2181,192.168.1.103:2182,192.168.1.103:2183
同理修改server3中的kafka配置，配置好之后，kafka集成環境已經配置完成，下面我們來驗證下集成環境是否OK。
驗證集群環境
依次先啟動zookeeper服務器
進入zookeeper的bin目錄，執行zkServer.cmd
依次啟動kafka服務器
進入kafka安裝目錄D:\bigData\kafka_2.11-1.1.0,按下shift+鼠標右鍵，選擇"在此處打開命令窗口"，打開命令行,在命令行中輸入：.\bin\windows\kafka-server-start.bat .\config\server.properties回車。正常啟動界面如圖：
在主機kafka服務器創建topic
主要參數說明：
partitions分區數
(1). partitions ：分區數，控制topic將分片成多少個log。可以顯示指定，如果不指定則會使用broker(server.properties)中的num.partitions配置的數量
(2). 雖然增加分區數可以提供kafka集群的吞吐量、但是過多的分區數或者或是單台服務器上的分區數過多，會增加不可用及延遲的風險。因為多的分區數，意味着需要打開更多的文件句柄、增加點到點的延時、增加客戶端的內存消耗。
(3). 分區數也限制了consumer的並行度，即限制了並行consumer消息的線程數不能大於分區數
(4). 分區數也限制了producer發送消息是指定的分區。如創建topic時分區設置為1，producer發送消息時通過自定義的分區方法指定分區為2或以上的數都會出錯的；這種情況可以通過alter –partitions 來增加分區數。
replication-factor副本
1. replication factor 控制消息保存在幾個broker(服務器)上，一般情況下等於broker的個數。
2. 如果沒有在創建時顯示指定或通過API向一個不存在的topic生產消息時會使用broker(server.properties)中的default.replication.factor配置的數量。
創建Topic
1. 創建主題，命名為"test20180430"，replicationfactor=3(因為這里有三個kafka服務器在運行)。可根據集群中kafka服務器個數來修改replicationfactor的數量，以便提高系統容錯性等。
2. 在D:\bigData\kafka_2.11-1.1.0\bin\windows目錄下打開新的命令行，輸入命令：kafka-topics.bat --create --zookeeper 192.168.1.130:2181 --replication-factor 3 --partitions 1 --topic test20180430回車。如圖：
查看Topic
kafka-topics.bat --zookeeper 192.168.1.130:2181 --describe --topic test20180430
結果說明:
第一行，列出了topic的名稱，分區數(PartitionCount),副本數(ReplicationFactor)以及其他的配置(Config.s)
Leader:1 表示為做為讀寫的broker的編號
Replicas:表示該topic的每個分區在那些borker中保存
Isr:表示當前有效的broker, Isr是Replicas的子集
創建Producer消息提供者
1. 在D:\bigData\kafka_2.11-1.1.0\bin\windows目錄下打開新的命令行，輸入命令：kafka-console-producer.bat --broker-list 192.168.1.130:9092,192.168.1.101:9093,192.168.1.101:9094 --topic test20180430 回車。(該窗口不要關閉)
創建Customer消息消費者
1. 在D:\bigData\kafka_2.11-1.1.0\bin\windows目錄下打開新的命令行，輸入命令：kafka-console-consumer.bat --zookeeper 192.168.1.130:2181 --topic test20180430回車。
以同樣的方式，將虛擬機上server2和server3的消息者customer創建，然后通過主機上的producer進行消息發送操作。
至此，集群環境已經驗證OK。
集群容錯性
1. 首先查看topic的信息
kafka-topics.bat --list --zookeeper 192.168.1.130:2181
查看指定的topic的詳細信息：
kafka-topics.bat --zookeeper 192.168.1.130:2181 --describe --topic test20180430
可以看到此時選舉的leader是0，即就是虛擬機中的kafka服務器，現在把虛擬機的kafka服務器給干掉。此時leader為變為1，消費者能繼續消費。
注意：zk的部署個數最好為基數，ZK集群的機制是只要超過半數的節點OK，集群就能正常提供服務。只有ZK節點掛得太多，只剩一半或不到一半節點能工作，集群才失效。