來源：https://www.cnblogs.com/guanzhyan/p/8974143.html

為了演示集群的效果，這里準(zhǔn)備一臺(tái)虛擬機(jī)(window 7)，在虛擬機(jī)中搭建了單IP多節(jié)點(diǎn)的zookeeper集群（多IP節(jié)點(diǎn)的也是同理的），并且在本機(jī)（win 7）和虛擬機(jī)中都安裝了kafka。

前期準(zhǔn)備說明：

1.三臺(tái)zookeeper服務(wù)器，本機(jī)安裝一個(gè)作為server1，虛擬機(jī)安裝兩個(gè)(單IP)

2.三臺(tái)kafka服務(wù)器，本機(jī)安裝一個(gè)作為server1，虛擬機(jī)安裝兩個(gè)。

備注：當(dāng)然你可以直接在虛擬機(jī)上安裝三個(gè)服務(wù)器分別為server1、server2、server3 。

• 虛擬機(jī)和本機(jī)網(wǎng)絡(luò)環(huán)境

1.將虛擬機(jī)的網(wǎng)絡(luò)模式調(diào)整為橋接模式，將虛擬機(jī)的防火墻功能關(guān)閉；

2.將主機(jī)的防火墻也關(guān)閉。

3.互相ping，測(cè)試是否能正常連接.。

• 下載kafka

從Kafka官網(wǎng)http://kafka./downloads下載Kafka安裝包。(要下載Binary downloads這個(gè)類型，不要下載源文件，方便使用)

• 下載zookeeper

從zookeeper官網(wǎng)http://zookeeper./releases.html下載zookeeper安裝包。

• 下載JDK

從官網(wǎng)http://www./download/下載(建議下載Oracle官方發(fā)布的Java),，配置環(huán)境變量。

zookeeper集群環(huán)境搭建

本機(jī)配置

修改本機(jī)sever1的zoo.cfg文件配置:

#存儲(chǔ)內(nèi)存中數(shù)據(jù)庫快照的位置，如果不設(shè)置參數(shù)，更新事務(wù)日志將被存儲(chǔ)到默認(rèn)位置。
#dataDir=/tmp/zookeeper
dataDir=D:/bigData/zookeeper-3.4.10/data
#日志文件的位置
dataLogDir=D:/bigData/zookeeper-3.4.10/zlog
#監(jiān)聽端口
clientPort=2181

#集群服務(wù)器配置
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:是一個(gè)數(shù)字，表示第幾號(hào)服務(wù)器

B:服務(wù)器IP地址

C:是一個(gè)端口號(hào)，用來集群成員的信息交換,表示這個(gè)服務(wù)器與集群中的leader服務(wù)器交換信息的端口

D:是在leader掛掉時(shí)專門用來進(jìn)行選舉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./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

創(chuàng)建serverID

在zoo.cfg配置文件中dataDir目錄中新建一個(gè)沒有后綴的myid文件，里邊寫1。

虛擬機(jī)配置

將本機(jī)的zookeeper安裝包拷貝到虛擬機(jī)上，然后在盤中新建兩個(gè)目錄，server2,server3目錄。

修改server2的zoo.cfg配置文件：

#存儲(chǔ)內(nèi)存中數(shù)據(jù)庫快照的位置，如果不設(shè)置參數(shù)，更新事務(wù)日志將被存儲(chǔ)到默認(rèn)位置。
#dataDir=/tmp/zookeeper
dataDir=E:/bigData/server2/zookeeper-3.4.10/data
#日志文件的位置
dataLogDir=E:/bigData/server2/zookeeper-3.4.10/zlog
#監(jiān)聽端口
clientPort=2182

#集群服務(wù)器配置
server.1=192.168.1.130:9000:7000
server.2=192.168.1.103:9001:7001
server.3=192.168.1.103:9002:7002

創(chuàng)建SeverID

在zoo.cfg配置文件中dataDir目錄中新建一個(gè)沒有后綴的myid文件，里邊寫2。

同理配置server3中的zookeeper，這樣zookeeper集群的配置工作已經(jīng)完成。

下面我們啟動(dòng)zookeeper集群：一個(gè)一個(gè)啟動(dòng)每一個(gè)zookeeper服務(wù)器：
先后進(jìn)入每一個(gè)zookeeper服務(wù)器的bin目錄下執(zhí)行zkserver.cmd,在先啟動(dòng)的zookeeper會(huì)打印其他zookeeper服務(wù)器未啟動(dòng)的錯(cuò)誤信息，直至到最后一個(gè)啟動(dòng)的時(shí)候就沒有異常信息了。正常界面：

zookeeper集群環(huán)境搭建遇到問題總結(jié):

1. zookeeper啟動(dòng)時(shí)報(bào)Cannot open channel to X at election address Error contacting service. It is probably not running.

解決方法：

a.     關(guān)閉所有服務(wù)器防火墻

b.    將本服務(wù)器的IP修改成0.0.0.0

例如：server.1=0.0.0.0:9000:7000

Kafka集群環(huán)境搭建：

下載kafka后解壓到某個(gè)目錄，然后修改server.properties配置。

server.properties修改說明：

broker.id=0
·    在kafka這個(gè)集群中的唯一標(biāo)識(shí)，且只能是正整數(shù)
port=9091
·    該服務(wù)監(jiān)聽的端口
host.name=192.168.1.130
·    broker 綁定的主機(jī)名稱(IP) 如果不設(shè)置將綁定所有的接口。
advertised.host.name=192.168.1.130
·    broker服務(wù)將通知消費(fèi)者和生產(chǎn)者 換言之，就是消費(fèi)者和生產(chǎn)者就是通過這個(gè)主機(jī)(IP)來進(jìn)行通信的。如果沒有設(shè)置就默認(rèn)采用host.name。
num.network.threads=2
·    broker處理消息的最大線程數(shù)，一般情況是CPU的核數(shù)
num.io.threads=8
·    broker處理IO的線程數(shù) 一般是num.network.threads的兩倍
socket.send.buffer.bytes=1048576
·    socket發(fā)送的緩沖區(qū)。socket調(diào)優(yōu)參數(shù)SO_SNDBUFF
socket.receive.buffer.bytes=1048576
·    socket接收的緩沖區(qū) socket的調(diào)優(yōu)參數(shù)SO_RCVBUF
socket.request.max.bytes=104857600
·    socket請(qǐng)求的最大數(shù)量，防止serverOOM。
log.dirs=\logs
·    kafka數(shù)據(jù)的存放地址，多個(gè)地址的話用逗號(hào)隔開。多個(gè)目錄分布在不同的磁盤上可以提高讀寫性能
num.partitions=2
·    每個(gè)tipic的默認(rèn)分區(qū)個(gè)數(shù)，在創(chuàng)建topic時(shí)可以重新制定
log.retention.hours=168
·    數(shù)據(jù)文件的保留時(shí)間 log.retention.minutes也是一個(gè)道理。
log.segment.bytes=536870912
·    topic中的最大文件的大小 -1表示沒有文件大小限制 log.segment.bytes 和log.retention.minutes 任意一個(gè) 
達(dá)到要求 都會(huì)刪除該文件 在創(chuàng)建topic時(shí)可以重新制定。若沒有.則選取該默認(rèn)值
log.retention.check.interval.ms=60000
·    文件大小檢查的周期時(shí)間，是否處罰 log.cleanup.policy中設(shè)置的策略
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
·    進(jìn)群鏈接時(shí)間超時(shí)

本機(jī)配置

修改本機(jī)server1的server.properties配置：

#在kafka這個(gè)集群中的唯一標(biāo)識(shí)，且只能是正整數(shù)
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 綁定的主機(jī)名稱(IP) 如果不設(shè)置將綁定所有的接口。
host.name=192.168.1.130

#kafka數(shù)據(jù)的存放地址，多個(gè)地址的話用逗號(hào)隔開。多個(gè)目錄分布在不同的磁盤上可以提高讀寫性能
# 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

#每個(gè)tipic的默認(rèn)分區(qū)個(gè)數(shù)，在創(chuàng)建topic時(shí)可以重新制定
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

虛擬機(jī)配置

將本機(jī)的kafka安裝包拷貝到虛擬機(jī)上，存放在server2,server3目錄。

注意：若本機(jī)拷貝到虛擬機(jī)中kafka之前有使用過，生成過topics，直接拷貝搭建集群，啟動(dòng)kafka時(shí)會(huì)報(bào)如下錯(cuò)誤：

原因：在單點(diǎn)環(huán)境下創(chuàng)建的topic中，kafka服務(wù)器的broker.id為0，在搭建集群時(shí)修改了borker.id，先前創(chuàng)建的topic就無法找到對(duì)應(yīng)的broker.id。

解決辦法：刪除原來的topic信息。

修改server2目錄中kafka配置：

#在kafka這個(gè)集群中的唯一標(biāo)識(shí)，且只能是正整數(shù)
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 綁定的主機(jī)名稱(IP) 如果不設(shè)置將綁定所有的接口。
host.name=192.168.1.103

#kafka數(shù)據(jù)的存放地址，多個(gè)地址的話用逗號(hào)隔開。多個(gè)目錄分布在不同的磁盤上可以提高讀寫性能
# 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

#每個(gè)tipic的默認(rèn)分區(qū)個(gè)數(shù)，在創(chuàng)建topic時(shí)可以重新制定
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集成環(huán)境已經(jīng)配置完成，下面我們來驗(yàn)證下集成環(huán)境是否OK。

驗(yàn)證集群環(huán)境

依次先啟動(dòng)zookeeper服務(wù)器

進(jìn)入zookeeper的bin目錄，執(zhí)行zkServer.cmd

依次啟動(dòng)kafka服務(wù)器

進(jìn)入kafka安裝目錄D:\bigData\kafka_2.11-1.1.0,按下shift+鼠標(biāo)右鍵，選擇"在此處打開命令窗口"，打開命令行,在命令行中輸入：.\bin\windows\kafka-server-start.bat .\config\server.properties回車。正常啟動(dòng)界面如圖：

在主機(jī)kafka服務(wù)器創(chuàng)建topic

主要參數(shù)說明：

partitions分區(qū)數(shù)

(1). partitions ：分區(qū)數(shù)，控制topic將分片成多少個(gè)log?？梢燥@示指定，如果不指定則會(huì)使用broker(server.properties)中的num.partitions配置的數(shù)量
(2). 雖然增加分區(qū)數(shù)可以提供kafka集群的吞吐量、但是過多的分區(qū)數(shù)或者或是單臺(tái)服務(wù)器上的分區(qū)數(shù)過多，會(huì)增加不可用及延遲的風(fēng)險(xiǎn)。因?yàn)槎嗟姆謪^(qū)數(shù)，意味著需要打開更多的文件句柄、增加點(diǎn)到點(diǎn)的延時(shí)、增加客戶端的內(nèi)存消耗。
(3). 分區(qū)數(shù)也限制了consumer的并行度，即限制了并行consumer消息的線程數(shù)不能大于分區(qū)數(shù)
(4). 分區(qū)數(shù)也限制了producer發(fā)送消息是指定的分區(qū)。如創(chuàng)建topic時(shí)分區(qū)設(shè)置為1，producer發(fā)送消息時(shí)通過自定義的分區(qū)方法指定分區(qū)為2或以上的數(shù)都會(huì)出錯(cuò)的；這種情況可以通過alter –partitions 來增加分區(qū)數(shù)。

replication-factor副本

1. replication factor 控制消息保存在幾個(gè)broker(服務(wù)器)上，一般情況下等于broker的個(gè)數(shù)。
2. 如果沒有在創(chuàng)建時(shí)顯示指定或通過API向一個(gè)不存在的topic生產(chǎn)消息時(shí)會(huì)使用broker(server.properties)中的default.replication.factor配置的數(shù)量。

創(chuàng)建Topic

1. 創(chuàng)建主題，命名為"test20180430"，replicationfactor=3(因?yàn)檫@里有三個(gè)kafka服務(wù)器在運(yùn)行)。可根據(jù)集群中kafka服務(wù)器個(gè)數(shù)來修改replicationfactor的數(shù)量，以便提高系統(tǒng)容錯(cuò)性等。
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

結(jié)果說明:
第一行，列出了topic的名稱，分區(qū)數(shù)(PartitionCount),副本數(shù)(ReplicationFactor)以及其他的配置(Config.s)
Leader:1 表示為做為讀寫的broker的編號(hào)
Replicas:表示該topic的每個(gè)分區(qū)在那些borker中保存
Isr:表示當(dāng)前有效的broker, Isr是Replicas的子集

創(chuàng)建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 回車。(該窗口不要關(guān)閉)

創(chuàng)建Customer消息消費(fèi)者

1. 在D:\bigData\kafka_2.11-1.1.0\bin\windows目錄下打開新的命令行，輸入命令：kafka-console-consumer.bat --zookeeper 192.168.1.130:2181 --topic test20180430回車。

以同樣的方式，將虛擬機(jī)上server2和server3的消息者customer創(chuàng)建，然后通過主機(jī)上的producer進(jìn)行消息發(fā)送操作。

至此，集群環(huán)境已經(jīng)驗(yàn)證OK。

集群容錯(cuò)性

1. 首先查看topic的信息
kafka-topics.bat --list --zookeeper 192.168.1.130:2181
查看指定的topic的詳細(xì)信息：
kafka-topics.bat --zookeeper 192.168.1.130:2181 --describe --topic test20180430

可以看到此時(shí)選舉的leader是0，即就是虛擬機(jī)中的kafka服務(wù)器，現(xiàn)在把虛擬機(jī)的kafka服務(wù)器給干掉。此時(shí)leader為變?yōu)?，消費(fèi)者能繼續(xù)消費(fèi)。

注意：zk的部署個(gè)數(shù)最好為基數(shù)，ZK集群的機(jī)制是只要超過半數(shù)的節(jié)點(diǎn)OK，集群就能正常提供服務(wù)。只有ZK節(jié)點(diǎn)掛得太多，只剩一半或不到一半節(jié)點(diǎn)能工作，集群才失效。