Kafka集群搭建及开机自启

集群硬件选择

版本号说明

kafka_2.11-2.2.0前面的2.11，2.12代表scala的版本号。实际版本号是2.2.0,分别为MajorVersion-MinorVersion-Patch号，Kafka下载地址https://kafka.apache.org/downloads

Zookeeper与kafka版本对应关系

Springboot与Kafka版本对应关系

Zookeeper集群安装

在192.28.4.35、4.37、4.43三天机器搭建一个最小集群（3个节点），一般都是奇数节点。

//解压zookeeper压缩包
tar -zxvf apache-zookeeper-3.4.14-bin.tar.gz

//
cp zoo.cfg.sample zoo.cfg
vim conf/zoo.cfg
如下图修改dataDir目录，在此目录下写入myid,添加集群通信配置

配置zookeeper开启自启动脚本

//其中ZOO_LOG_DIR目录需要自己手动建立
[Unit]
Description=zookeeper.service
After=network.target
[Service]
Type=forking
Environment=ZOO_LOG_DIR=/usr/local/kafka_zookeeper/zookeeper_log
ExecStart=/usr/local/kafka_zookeeper/zookeeper-3.4.14/bin/zkServer.sh start
ExecStop=/usr/local/kafka_zookeeper/zookeeper-3.4.14/bin/zkServer.sh stop
ExecReload=/usr/local/kafka_zookeeper/zookeeper-3.4.14/bin/zkServer.sh restart
Restart=always
[Install]
WantedBy=multi-user.target

查看zookeeper运行状态

使用ZooKeeper的四字命令，你可以直接通过telnet或者nc命令查看状态。

常用的四字命令如下：

conf: 打印ZooKeeper的配置信息
cons: 列出所有的客户端会话链接
crst: 重置所有的客户端连接
dump: 打印集群的所有会话信息，包括ID，以及临时节点等信息。用在Leader节点上才有效果。
envi: 列出所有的环境参数
ruok: “谐音为Are you ok”。检查当前服务器是否正在运行。
stat: 获取ZooKeeper服务器运行时的状态信息，包括版本，运行时角色，集群节点个数等信息。
srst: 重置服务器统计信息
srvr: 和stat输出信息一样，只不过少了客户端连接信息。
wchs: 输出当前服务器上管理的Watcher概要信息
wchc: 输出当前服务器上管理的Watcher的详细信息，以session为单位进行归组
wchp: 和wchc非常相似，但是以节点路径进行归组
mntr: 输出比stat更为详细的服务器统计信息

Kafka集群搭建

这里同样使用192.28.4.35、4.37、4.43三条搭建kafka集群。修改每台上面kafka的server.properties配置。

broker.id=35 #不能重复的整数
listeners=PLAINTEXT://192.28.4.35:9092 #添加此配置
log.dirs=/usr/local/kafka_zookeeper/kafka_log #修改日志文件地址
zookeeper.connect=192.28.4.35:2181,192.28.4.37:2181,192.28.4.43:2181 #配置zookeeper地址
auto.create.topics.enable=false
auto.leader.rebanlance.enable=false #禁止自动选举
#设置可以删除topic
delete.topic.enable=true
#设置成功写入某条信息必须要等待相应完成的ISR副本数，需要配合acks=all或者acks=-1 如果有三个副本那么至少要两个写入成功才可以
min.insync.replicas=2
#延时调优
#从源broker复制消息的提取线程数，递增该值可提高follower brokeer的I/O并发 <=cpu核数
num.replica.fetchers=4
#高可用性设置
#即使ISR全部宕机，还有非ISR可以作为leader,这里防止脏数据设置为false
unclean.leader.eclection.enable=false
#成功写入消息至少要等1个副本同步完成
min.insync.replicas=1
#broker可接收的最大字节数6M
message.max.bytes = 62914560
#kafka可复制的消息的最大字节数，要比message.max.bytes大
replica.fetch.max.bytes = 104857600
#12小时
log.retention.hours=12

bin/kafka-server-start.sh config/server.properties #修改完3台后启动kafka测试下

Kafka自启脚本

[Unit]
Description=kafka service
Requires=network.target zookeeper.service
After=network.target zookeeper.service

[Service]
Type=simple
ExecStart=/usr/local/kafka_zookeeper/kafka_2.12-2.2.2/bin/kafka-server-start.sh  /usr/local/kafka_zookeeper/kafka_2.12-2.2.2/config/server.properties
ExecStop=/usr/local/kafka_zookeeper/kafka_2.12-2.2.2/bin/kafka-server-stop.sh
Restart=on-failure
RestartSec=60

[Install]
WantedBy=multi-user.target