Zookeeper&Kafka部署

1.1zookeeper单节点安装

1.1.1.获取apache-zookeeper-3.6.0-bin.zip安装包

1.1.2.上传安装包至~/software安装目录下执行如下命令解压。

unzip apache-zookeeper-3.6.0-bin.zip

1.1.3.修改安装目录下配置文件conf/zoo.cfg

1.1.4.日志配置（可选）

此配置为可选配置，各项目可根据需求选择是否配置。

日志按照日期存放

进入zookeeper的conf目录下，打开log4j.properties文件，做如下修改使zookeeper日志可按照日期存放。

#将ZOO_LOG4J_PROP="INFO,CONSOLE"修改为：
ZOO_LOG4J_PROP="INFO,ROLLINGFILE"

  #log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender ### 日志按大小轮循 ###
  #log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold}
  #log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file}
log4j.appender.ROLLINGFILE=org.apache.log4j.DailyRollingFileAppender ### 日志按时间轮循 ###
log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold}
log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file}
log4j.appender.ROLLINGFILE.DatePattern='.'yyyy-MM-dd  ###日志名称后拼接的日期格式（按天存放）
#log4j.appender.ROLLINGFILE.MaxFileSize=${zookeeper.log.maxfilesize}
#log4j.appender.ROLLINGFILE.MaxBackupIndex=${zookeeper.log.maxbackupindex}

进入zookeeper的bin目录下，打开 zkEnv.sh文件，修改为：

if [ "x${ZOO_LOG4J_PROP}" = "x" ]
then
  #ZOO_LOG4J_PROP="INFO,CONSOLE"
ZOO_LOG4J_PROP="INFO,ROLLINGFILE"
fi

上述修改完成之后，重启zookeeper之后zookeeper的运行日志就会按照日期存放。

脚本定期清理zk快照和日志文件

正常运行过程中，ZK会不断地把快照数据和事务日志输出到data和log这两个目录，并且如果没有人为操作的话，ZK自己是不会清理这些文件的。 我们可以采用脚本切割。

#!/bin/bash
#清理zookeeper的日志 保留7天内的
#
#datadir=/opt/zookeeper/zkdata/version-2
logdir=/app/dcits/software/apache-zookeeper-3.6.0-bin/logs
#cd $datadir
#find . -mtime +7 |xargs rm -rf 
cd  $logdir
find . -mtime +7 |xargs rm -rf

注意：如果仅仅是清理快照日志数据：在zoo.cfg中配置即可

autopurge.snapRetainCount=20   #保留日志快照数量

autopurge.purgeInterval=48     #保留48小时内的日志数据

1.1.5.启动服务

进入zookeeper的bin目录，执行启动命令：

./zkServer.sh start

1.1.6.检查服务

进入zookeeper的bin目录，执行启动命令：

./zkServer.sh status

1.2.zookeeper集群安装

**说明：在Zookeeper集群中，若超过半数以上服务节点不可用，才会造成整个服务不可用，所以其集群节点数一般都是至少3个节点以上的奇数个 **

1.2.1.创建myId文件

注意：myid文件需创建在zkdata（dataDir配置项的路径）目录下，其值为后续zoo.cfg中配置的第几号服务器 （server.X）

1.2.2. zoo.cfg配置文件新增如下内容：

注意：所有Zookeeper集群中的节点机器，除了myid文件中的值不一样外，其它配置均一样

1.3.kafka单节点部署

注：kafka需要zookeeper的支持，需优先启动zookeeper。

1.3.1.上传安装包至安装目录下

​

1.3.2.解压安装包

unzip kafka_2.12-2.4.1.zip

1.3.3.修改配置文件config/server.properties

(参照182服务器/app/dcits/software/kafka_2.12-2.4.1/config/server.properties)

（1）将listeners修改为安装kafka的主机IP地址

（2）将zookeeper.connect修改为安装ZooKeeper的主机IP地址。如果ZK部署为集群模式，则配置文件格式如：10.139.*.:2181,10.139.*. :2181,10.139.4*.:2181*。

（3）修改kafka消息存放路径，默认为/tmp/kafka-logs路径

（4）修改kafka运行日志输出配置（根据需求可选择是否修改配置）

kafka中默认运行日志的输出文件是小时级别的，简单来说就是一个小时一个。

对于一小时一个日志文件，可能在查看日志文件排查问题的时候比较麻烦，故可以将其修改为一天一个日志文件即可。

进入kafka安装路径下的config文件夹下，按照如下方式修改log4j.properties文件：

log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH   #一小时一个
log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd      #一天一个

（5）kafka数据清理配置（根据需求可选择是否配置）

进入kafka安装路径下的config打开 server.properties，修改如下参数：

#kafka提供的清理策略有两种，一种是直接删除，一种是压缩
(1)删除
log.cleanup.policy=delete   #启用删除策略
#直接删除，删除后的消息不可恢复。可配置以下两个策略：
#清理超过指定时间清理：  
log.retention.hours=16
#超过指定大小后，删除旧的消息：
log.retention.bytes=1073741824
#log.retention.bytes为topic每个分区的最大文件大小，一个topic的大小限制 =分区数*log.retention.bytes。-1没有大小限制，log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖
#为了避免在删除时阻塞读操作，采用了copy-on-write形式的实现，删除操作进行时，读取操作的二分查找功能实际是在一个静态的快照副本上进行的，这类似于Java的CopyOnWriteArrayList。

（2）#将数据压缩，只保留每个key最后一个版本的数据。
#首先在broker的配置中设置
log.cleaner.enable=true #启用cleaner，这个默认是关闭的
#在topic的配置中设置
log.cleanup.policy=compact #启用压缩策略。

1.3.4.启动服务

提供以下两种启动方式：

（1）进入bin目录下，执行如下命令

./kafka-server-start.sh -daemon ../config/server.properties

（2）进入kafka安装目录下，执行如下命令

./start.sh

1.3.5.校验服务

执行命令netstat -an | grep PORT，端口被占用(LISTEN)表示启动成功。

1.3.6.停止服务

提供两种停止服务的方法

（1）进入bin路径下，执行如下命令

./kafka-server-stop.sh

（2）进入kafka安装目录下，执行如下命令

./stop.sh

1.4.kafka集群部署

注意： 在单机的基础之上，只要修改server.properties配置文件中的broker.id是其在集群能够

保证唯一即可， kafka集群中的节点没有主从之分。

Client 端：配置server节点的ip端口，多个地址之间用逗号分开。