kafka集群搭建部署测试

由于需求，需要将单节点kafka扩展成为kafka集群，中间遇到了很多问题，特将最终的正确步骤记录下来，仅作参考。

1.软件环境（ubuntu server 3台）

1.1.1.1 server1

2.2.2.2 server2

3.3.3.3 server3

注意机器个数为（2*n+1），同时zookeeper需要java环境才能运行，首先在三台机器上都配置java环境。

java环境配置:

a).下载jdk-8u151-linux-x64.tar.gz至/home/ubuntu

b).tar -xvf jdk-8u151-linux-x64.tar.gz

c)sudo vim /etc/profile,在文件末尾添加

export JAVA_HOME=/home/ubuntu/jdk1.8.0_151
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

d)source /etc/profile

2.搭建zookeeper集群

a).下载zookeeper-3.3.6.tar.gz至/opt/zookeeper中
b).tar -xvf zookeeper-3.3.6.tar.gz
c).sudo mkdir zookeeper #项目目录
sudo mkdir zkdata #存放快照日志
sudo mkdir zkdatalog#存放事物日志
d).cd conf
cp zoo_sample.cfg zoo.cfg
sudo vim zoo.cfg
配置项如下：
#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.
# the port at which the clients will connect
dataDir=/opt/zookeeper/zkdata
dataLogDir=/opt/zookeeper/zkdatalog
clientPort=2181
server.1=1.1.1.1:2888:3888
server.2=2.2.2.2:2888:3888
server.3=3.3.3.3:2888:3888
三台集群均做相同处理后，使用如下命令在每台机器上生成相应的id文件
#server1
echo "1" > /opt/zookeeper/zkdata/myid
#server2
echo "2" > /opt/zookeeper/zkdata/myid
#server3
echo "3" > /opt/zookeeper/zkdata/myid
接下来启动zookeeper集群（3台均需启动）
./bin/zkServer.sh start ../conf/zoo.cfg
启动后调用
./bin/zkServer.sh status 可以查看节点的模式，分为follower与leader
至此，zookeeper集群配置完毕

3.搭建kafka集群

mkdir /opt/kafka
mkdir /opt/kafkalogs
将kafka_2.11-1.0.0.tgz下载至/opt/kafka
tar -xvf kafka_2.11-1.0.0.tgz
cd kafka_2.11-1.0.0
vim /config/server.properties
配置项如下:
broker.id=1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka/kafkalogs
num.partitions=8
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=0.0.0.0:2181,2.2.2.2:2181,3.3.3.3:2181
advertised.host.name=1.1.1.1
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
需要注意的地方有：
broker.id三台机器不可重复，作为唯一标识
log.dirs配置为我们之前创建的kafkalogs文件夹，不要使用默认配置的/tmp临时文件夹，否则机器重启会导致数据丢失
advertised.host.name=1.1.1.1必须配置否则会导致生产与消费数据的失败
配置完毕后启动kafka集群
./bin/kafka-server-start.sh ../config/server.properties (三台机器)
测试创建topic
./kafka-topics.sh --create --zookeeper 1.1.1.1:2181 --replication-factor 2 --partitions 1 --topic shuaige
#解释
--replication-factor 2   #复制两份
--partitions 1 #创建1个分区
--topic #主题为shuaige
'''在一台服务器上创建一个发布者'''
#创建一个broker，发布者
./kafka-console-producer.sh --broker-list 1.1.1.1:19092 --topic test
'''在一台服务器上创建一个订阅者'''
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

3.kafka集群监控的搭建

1、获取kafka-manager源码，并编译打包
# cd /usr/local
# git clone https://github.com/yahoo/kafka-manager # cd kafka-manager
# ./sbt clean dist
2.编译成功后，会在target/universal下生成一个zip包
# cd /usr/local/kafka-manager/target/universal
# unzip kafka-manager-1.3.3.7.zip
将application.conf中的kafka-manager.zkhosts的值设置为你的zk地址
如：kafka-manager.zkhosts="1.1.1.1:2181,2.2.2.2:2181,3.3.3.3:2181"
3.后台运行
默认端口（9000）
# ./kafka-manager -h
# nohup ./kafka-manager -Dconfig.file=../conf/application.conf &
指定端口
# nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9001 &
接下来便可以通过http://1.1.1.1:9000/来添加自己的集群，从而进一步管理