Zookeeper+Kafka完全分布式实战部署
作者：尹正杰
版权声明：原创作品，谢绝转载！否则将追究法律责任。
其实我之前部署过kafak和zookeeper的完全分布式，集群是可以正常使用没错，但是在调优方案我做的很少，本次部署模拟我实际生成环境中的kafka版本zookeeper的一些调优措施，以及一些脚本管理等。部署集群需要你自行安装jdk，本篇博客就直接上干货了。
关于本篇博客的测试版本视频：链接：https://pan.baidu.com/s/1S3UqwTH05RKQOuQ9bwOFMg 提取码：jsv3
关于kafka操作系统的优化，可参考：https://www.cnblogs.com/yinzhengjie/p/9993719.html
一.集群的调优方向
1>.调大zookeeper的heap内存，默认是1G，可以根据服务器大小配置其堆内存为2G或者4G足矣(kafka实时传输的数据如果达到PB级别的话，得观察一下YGC和FGC的值可以适当再次调大)；
2>.修改kafka的副本数，默认的副本数是1，建议修改为2，如果副本数为2，那么容灾能力就是1，如果副本数3，则容灾能力就是2，当然副本数越多，可能会导致集群的性能下降，但是可靠性更强，各有利弊，我这里推荐副本数为2；
3>.kafka推荐分区数，默认的分区数是1，理论上来说，parition的数量小于core的数量的话，值越大，kafka的吞吐量就越高，但是你必须得考虑你的磁盘IO的瓶颈，因此我不推荐你将分区数这只过大，我建议这个值大于broker的数量，比如我的集群broker的只有5台，我的集群的partition数量是20；
4>.kafka的heap内存，默认也是1G，生成环境中建议将它调大，不知道大家有没有发现，你broker的heap内存不管有多的，它都能给你吃满！我在生成环境中给kafka的heap内存是6G(kafka主要使用堆外内存，即大量使用操作系统的页缓存，因此其并不需要分配太多的堆内存空间)，zookeeper给的是2G，剩下的全部给操作系统预留着，否则你的机器会非常的卡顿；
5>.各种kafka配置文件调优，我这里就不一一赘述了，我在本文中已经有详细的介绍。
二.部署zookeeper集群
1>.下载截止20181110日最新稳定版本版本的zookeeper-3.4.13.tar.gz（2018-07-16 ）版本
[root@yinzhengjie zookeeper]# wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
---- ::-- https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
Resolving mirrors.tuna.tsinghua.edu.cn (mirrors.tuna.tsinghua.edu.cn)... 101.6.8.193, :f000:::::
Connecting to mirrors.tuna.tsinghua.edu.cn (mirrors.tuna.tsinghua.edu.cn)|101.6.8.193|:... connected.
HTTP request sent, awaiting response... OK
Length: (35M) [application/x-gzip]
Saving to: ‘zookeeper-3.4..tar.gz’ %[====================================================================================================================================================================================================================================>] ,, .5MB/s in .4s -- :: (81.5 MB/s) - ‘zookeeper-3.4..tar.gz’ saved [/] [root@yinzhengjie zookeeper]#
[root@yinzhengjie zookeeper]# ll
total
-rw-r--r-- root root Jul : zookeeper-3.4..tar.gz
[root@yinzhengjie zookeeper]#
[root@yinzhengjie zookeeper]# wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
2>.解压zookeeper
[root@yinzhengjie zookeeper]# tar -zxf zookeeper-3.4..tar.gz -C /soft/
3>.创建软连接
[root@yinzhengjie ~]# ln -s /soft/zookeeper-3.4./ /soft/zk
4>.创建配置zookeeper的堆内存配置文件
[root@yinzhengjie ~]# cat /soft/zk/conf/java.env
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com #指定JDK的安装路径
export JAVA_HOME=/soft/jdk #指定zookeeper的heap内存大小
export JVMFLAGS="-Xms2048m -Xmx2048m $JVMFLAGS"
[root@yinzhengjie ~]#
5>.修改zookeeper的配置文件zoo.cfg（需要手动创建，或者从“ /soft/zk/conf/zoo_sample.cfg ”赋值一个模板即可）
[root@yinzhengjie ~]# cat /soft/zk/conf/zoo.cfg
# 滴答，计时的基本单位，默认是2000毫秒，即2秒。它是zookeeper最小的时间单位，用于丈量心跳时间和超时时间等，通常设置成默认2秒即可。
tickTime= # 初始化限制是10滴答，默认是10个滴答，即默认是20秒。指定follower节点初始化是链接leader节点的最大tick次数。
initLimit=5 # 数据同步的时间限制，默认是5个滴答，即默认时间是10秒。设定了follower节点与leader节点进行同步的最大时间。与initLimit类似，它也是以tickTime为单位进行指定的。
syncLimit=2 # 指定zookeeper的工作目录，这是一个非常重要的参数，zookeeper会在内存中在内存只能中保存系统快照，并定期写入该路径指定的文件夹中。生产环境中需要注意该文件夹的磁盘占用情况。
dataDir=/home/yinzhengjie/zookeeper # 监听zookeeper的默认端口。zookeeper监听客户端链接的端口，一般设置成默认2181即可。
clientPort= # 这个操作将限制连接到 ZooKeeper 的客户端的数量，限制并发连接的数量，它通过 IP 来区分不同的客户端。此配置选项可以用来阻止某些类别的 Dos 攻击。将它设置为 或者忽略而不进行设置将会取消对并发连接的限制。
#maxClientCnxns= # 在上文中已经提到，3.4.0及之后版本，ZK提供了自动清理事务日志和快照文件的功能，这个参数指定了清理频率，单位是小时，需要配置一个1或更大的整数，默认是0，表示不开启自动清理功能。
#autopurge.purgeInterval= # 这个参数和上面的参数搭配使用，这个参数指定了需要保留的文件数目。默认是保留3个。
#autopurge.snapRetainCount= #server.x=[hostname]:nnnnn[:nnnnn]，这里的x是一个数字，与myid文件中的id是一致的。右边可以配置两个端口，第一个端口用于F和L之间的数据同步和其它通信，第二个端口用于Leader选举过程中投票通信。
server.=10.1.3.117::
server.=10.1.3.118::
server.=10.1.3.119::
[root@yinzhengjie ~]#
6>.编写zookeeper的启动脚本
[root@yinzhengjie ~]# cat /usr/local/bin/xzk.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com #判断用户是否传参
if [ $# -ne ];then
echo "无效参数，用法为: $0 {start|stop|restart|status}"
exit
fi #获取用户输入的命令
cmd=$ #定义函数功能
function zookeeperManger(){
case $cmd in
start)
echo "启动服务"
remoteExecution start
;;
stop)
echo "停止服务"
remoteExecution stop
;;
restart)
echo "重启服务"
remoteExecution restart
;;
status)
echo "查看状态"
remoteExecution status
;;
*)
echo "无效参数，用法为: $0 {start|stop|restart|status}"
;;
esac
} #定义执行的命令
function remoteExecution(){
for (( i= ; i<= ; i++ )) ; do
tput setaf
echo ========== kafka${i}.aggrx zkServer.sh $ ================
tput setaf
ssh kafka${i}.aggrx "source /etc/profile ; zkServer.sh $1"
done
} #调用函数
zookeeperManger
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# chmod +x /usr/local/bin/xzk.sh
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# ll /usr/local/bin/xzk.sh
-rwxr-xr-x root root Nov : /usr/local/bin/xzk.sh
[root@yinzhengjie ~]#
8>.同步系统配置文件
[root@yinzhengjie ~]# cat /usr/local/bin/xrsync.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com #判断用户是否传参
if [ $# -lt ];then
echo "请输入参数";
exit
fi #获取文件路径
file=$@ #获取子路径
filename=`basename $file` #获取父路径
dirpath=`dirname $file` #获取完整路径
cd $dirpath
fullpath=`pwd -P` #同步文件到Kafka集群
for (( i=;i<=;i++ ))
do
#使终端变绿色
tput setaf
echo =========== kafka${i}.aggrx : $file ===========
#使终端变回原来的颜色，即白灰色
tput setaf
#远程执行命令
rsync -lr $filename `whoami`@kafka${i}.aggrx:$fullpath
#判断命令是否执行成功
if [ $? == ];then
echo "命令执行成功"
fi
done
[root@yinzhengjie ~]#
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# ll /usr/local/bin/xrsync.sh
-rwxr-xr-x root root Oct : /usr/local/bin/xrsync.sh
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# cat /usr/local/bin/xrsync.sh #用于分发配置文件，需要你手动配置SSH无秘钥登录
[root@yinzhengjie ~]# tail - /etc/profile
#ADD Zookeeper PATH BY yinzhengjie
ZOOKEEPER=/soft/zk
PATH=$PATH:$ZOOKEEPER/bin
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# xrsync.sh /etc/profile
=========== yinzhengjie.aggrx : /etc/profile ===========
命令执行成功
=========== kafka117.aggrx : /etc/profile ===========
命令执行成功
=========== kafka118.aggrx : /etc/profile ===========
命令执行成功
=========== kafka119.aggrx : /etc/profile ===========
命令执行成功
=========== kafka120.aggrx : /etc/profile ===========
命令执行成功
[root@yinzhengjie ~]#
9>.将上述解压的配置文件使用xrsync.sh同步到其它节点
注意，接下来需要在“/home/yinzhengjie/zookeeper/”目录中创建一个myid，并写入配置文件。也可以使用一个shell循环搞定，仅供参考。
[root@yinzhengjie ~]# for (( i=;i<=;i++ )) do ssh kafka${i}.aggrx "echo -n $i > /home/yinzhengjie/zookeeper/myid" ;done
10>.启动zookeeper并查看状态
[root@yinzhengjie ~]# xzk.sh status #这是查看zookeeper的状态，如果是启动，或停止zookeeper，直接调用start或者stop方法即可
查看状态
========== kafka117.aggrx zkServer.sh status ================
ZooKeeper JMX enabled by default
Using config: /soft/zk/bin/../conf/zoo.cfg
Mode: follower
========== kafka118.aggrx zkServer.sh status ================
ZooKeeper JMX enabled by default
Using config: /soft/zk/bin/../conf/zoo.cfg
Mode: leader #很显然，该节点为zookeeper节点。
========== kafka119.aggrx zkServer.sh status ================
ZooKeeper JMX enabled by default
Using config: /soft/zk/bin/../conf/zoo.cfg
Mode: follower
[root@yinzhengjie ~]#
三.部署kafka集群
1>.官网下载kafka（kafka_2.11-0.10.2.1.tgz）
[root@yinzhengjie ~]# wget https://archive.apache.org/dist/kafka/0.10.2.1/kafka_2.11-0.10.2.1.tgz
---- ::-- https://archive.apache.org/dist/kafka/0.10.2.1/kafka_2.11-0.10.2.1.tgz
Resolving archive.apache.org (archive.apache.org)... 163.172.17.199
Connecting to archive.apache.org (archive.apache.org)|163.172.17.199|:... connected.
HTTP request sent, awaiting response... OK
Length: (36M) [application/x-gzip]
Saving to: ‘kafka_2.-0.10.2.1.tgz’ %[====================================================================================================================================================================================================================================>] ,, 228KB/s in 2m 58s -- :: ( KB/s) - ‘kafka_2.-0.10.2.1.tgz’ saved [/] [root@yinzhengjie ~]#
[root@yinzhengjie ~]# ll
total
-rw-r--r-- root root Apr kafka_2.-0.10.2.1.tgz
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# wget https://archive.apache.org/dist/kafka/0.10.2.1/kafka_2.11-0.10.2.1.tgz
2>.解压kafka
[root@yinzhengjie ~]# tar -zxf kafka_2.-0.10.2.1.tgz -C /soft/
3>.创建软连接
[root@yinzhengjie ~]# ln -s /soft/kafka_2.-0.10.2.1/ /soft/kafka
4>.修改kafka的配置文件（server.properties）
[root@yinzhengjie ~]# cat /soft/kafka/config/server.properties | grep -v ^# | grep -v ^$
broker.id=
delete.topic.enable=true
auto.create.topics.enable=false
port=
host.name=10.1.3.116
num.network.threads=
num.io.threads=
socket.send.buffer.bytes=
socket.receive.buffer.bytes=
socket.request.max.bytes=
queued.max.requests=
log.dirs=/home/yinzhengjie/kafka/logs,/home/yinzhengjie/kafka/logs2,/home/yinzhengjie/kafka/log3
num.partitions=
num.recovery.threads.per.data.dir=
default.replication.factor=
message.max.bytes=
log.retention.hours=
log.segment.bytes=
log.retention.check.interval.ms=
zookeeper.connect=10.1.3.117:,10.1.3.118:,10.1.3.119:
zookeeper.session.timeout.ms=
zookeeper.connection.timeout.ms=
max.request.size=
fetch.message.max.bytes=
replica.fetch.max.bytes=
replica.fetch.wait.max.ms=
unclean.leader.election=false
num.replica.fetchers=
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# cat /soft/kafka/config/server.properties
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License. # see kafka.server.KafkaConfig for additional details and defaults ############################# Server Basics ############################# #每一个broker在集群中的唯一表示，要求是正数。当该服务器的IP地址发生改变时，broker.id没有变化，则不会影响consumers的消息情况
broker.id= #这就是说，这条命令其实并不执行删除动作，仅仅是在zookeeper上标记该topic要被删除而已，同时也提醒用户一定要提前打开delete.topic.enable开关，否则删除动作是不会执行的。
delete.topic.enable=true #是否允许自动创建topic，若是false，就需要通过命令创建topic
auto.create.topics.enable=false ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092 #Socket服务器侦听的地址。如果没有配置，它将获得从Java.NET.InAddio.GETCANONICALITHAMEMENE（）返回的值
#listeners=PLAINTEXT://10.1.3.116:9092 #broker server服务端口
port= #broker的主机地址，若是设置了，那么会绑定到这个地址上，若是没有，会绑定到所有的接口上，并将其中之一发送到ZK，一般不设置
host.name=10.1.3.116
# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName(). #kafka 0.9.x以后的版本新增了advertised.listeners配置,kafka 0.9.x以后的版本不要使用 advertised.host.name 和 advertised.host.port 已经deprecated.如果配置的话，它使用 "listeners" 的值。否则，它将使用从java.net.InetAddress.getCanonicalHostName()返回的值。
#advertised.listeners=PLAINTEXT://your.host.name:9092 #将侦听器(listener)名称映射到安全协议，默认情况下它们是相同的。有关详细信息，请参阅配置文档。
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL #处理网络请求的最大线程数
num.network.threads= #处理磁盘I/O的线程数
num.io.threads= #套接字服务器使用的发送缓冲区（SOYSNDBUF)
socket.send.buffer.bytes= #套接字服务器使用的接收缓冲区（SOYRCVBUF）
socket.receive.buffer.bytes= #套接字服务器将接受的请求的最大大小（对OOM的保护）
socket.request.max.bytes= #I/O线程等待队列中的最大的请求数，超过这个数量，network线程就不会再接收一个新的请求。应该是一种自我保护机制。
queued.max.requests= ############################# Log Basics ############################# #日志存放目录，多个目录使用逗号分割,如果你有多块磁盘，建议配置成多个目录，从而达到I/O的效率的提升。
log.dirs=/home/yinzhengjie/kafka/logs,/home/yinzhengjie/kafka/logs2,/home/yinzhengjie/kafka/logs3 #每个topic的分区个数，若是在topic创建时候没有指定的话会被topic创建时的指定参数覆盖
num.partitions= #在启动时恢复日志和关闭时刷盘日志时每个数据目录的线程的数量，默认1
num.recovery.threads.per.data.dir= # 默认副本数
default.replication.factor= #服务器接受单个消息的最大大小，即消息体的最大大小，单位是字节
message.max.bytes= # 自动负载均衡,如果设为true，复制控制器会周期性的自动尝试，为所有的broker的每个partition平衡leadership，为更优先(preferred)的replica分配leadership。
# auto.leader.rebalance.enable=false ############################# Log Flush Policy ############################# #在强制fsync一个partition的log文件之前暂存的消息数量。调低这个值会更频繁的sync数据到磁盘，影响性能。通常建议人家使用replication来确保持久性，而不是依靠单机上的fsync，但是这可以带来更多的可靠性，默认10000。
#log.flush.interval.messages= #2次fsync调用之间最大的时间间隔，单位为ms。即使log.flush.interval.messages没有达到，只要这个时间到了也需要调用fsync。默认3000ms.
#log.flush.interval.ms= ############################# Log Retention Policy ############################# # 日志保存时间 (hours|minutes)，默认为7天（168小时）。超过这个时间会根据policy处理数据。bytes和minutes无论哪个先达到都会触发。
log.retention.hours= #日志数据存储的最大字节数。超过这个时间会根据policy处理数据。
#log.retention.bytes= #控制日志segment文件的大小，超出该大小则追加到一个新的日志segment文件中（-1表示没有限制）
log.segment.bytes= # 当达到下面时间，会强制新建一个segment
#log.roll.hours = * # 日志片段文件的检查周期，查看它们是否达到了删除策略的设置（log.retention.hours或log.retention.bytes）
log.retention.check.interval.ms= #是否开启压缩
#log.cleaner.enable=false #日志清理策略选择有：delete和compact主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖
#log.cleanup.policy=delete # 日志压缩运行的线程数
#log.cleaner.threads= # 压缩的日志保留的最长时间
#log.cleaner.delete.retention.ms= ############################# Zookeeper ############################# #zookeeper集群的地址，可以是多个，多个之间用逗号分割.
zookeeper.connect=10.1.3.117:,10.1.3.118:,10.1.3.119: #ZooKeeper的最大超时时间，就是心跳的间隔，若是没有反映，那么认为已经死了，不易过大
zookeeper.session.timeout.ms= #指定多久消费者更新offset到zookeeper中。注意offset更新时基于time而不是每次获得的消息。一旦在更新zookeeper发生异常并重启，将可能拿到已拿到过的消息,连接zk的超时时间
zookeeper.connection.timeout.ms= #请求的最大大小为字节,请求的最大字节数。这也是对最大记录尺寸的有效覆盖。注意：server具有自己对消息记录尺寸的覆盖，这些尺寸和这个设置不同。此项设置将会限制producer每次批量发送请求的数目，以防发出巨量的请求。
max.request.size= #每次fetch请求中，针对每次fetch消息的最大字节数。这些字节将会督导用于每个partition的内存中，因此，此设置将会控制consumer所使用的memory大小。这个fetch请求尺寸必须至少和server允许的最大消息尺寸相等，否则，producer可能发送的消息尺寸大于consumer所能消耗的尺寸。
fetch.message.max.bytes= #ZooKeeper集群中leader和follower之间的同步时间,换句话说：一个ZK follower能落后leader多久。
#zookeeper.sync.time.ms= ############################# Replica Basics ############################# # leader接收follower的"fetch请求"的超时时间,默认是10秒。
# replica.lag.time.max.ms= # 如果relicas落后太多,将会认为此partition relicas已经失效。而一般情况下,因为网络延迟等原因,总会导致replicas中消息同步滞后。如果消息严重滞后,leader将认为此relicas网络延迟较大或者消息吞吐能力有限。在broker数量较少,或者网络不足的环境中,建议提高此值.follower落后于leader的最大message数,这个参数是broker全局的。设置太大 了，影响真正“落后”follower的移除;设置的太小了，导致follower的频繁进出。无法给定一个合适的replica.lag.max.messages的值,因此不推荐使用，据说新版本的Kafka移除了这个参数。
#replica.lag.max.messages= # follower与leader之间的socket超时时间
#replica.socket.timeout.ms= # follower每次fetch数据的最大尺寸
replica.fetch.max.bytes= # follower的fetch请求超时重发时间
replica.fetch.wait.max.ms= # fetch的最小数据尺寸
#replica.fetch.min.bytes= # 是否允许控制器关闭broker ,默认值为true,它会关闭所有在这个broker上的leader，并转移到其他broker，建议启用，增加集群稳定性。
# controlled.shutdown.enable = false #0.11..0版本开始unclean.leader.election.enable参数的默认值由原来的true改为false，可以关闭unclean leader election，也就是不在ISR(IN-Sync Replica)列表中的replica，不会被提升为新的leader partition。kafka集群的持久化力大于可用性，如果ISR中没有其它的replica，会导致这个partition不能读写。
unclean.leader.election=false # follower中开启的fetcher线程数, 同步速度与系统负载均衡
num.replica.fetchers= # partition leader与replicas之间通讯时,socket的超时时间
#controller.socket.timeout.ms= # partition leader与replicas数据同步时,消息的队列尺寸.
#controller.message.queue.size= #指定将使用哪个版本的 inter-broker 协议。 在所有经纪人升级到新版本之后，这通常会受到冲击。升级时要设置
#inter.broker.protocol.version=0.10. #指定broker将用于将消息添加到日志文件的消息格式版本。 该值应该是有效的ApiVersion。 一些例子是：0.8.，0.9.0.0，0.10.。 通过设置特定的消息格式版本，用户保证磁盘上的所有现有消息都小于或等于指定的版本。 不正确地设置这个值将导致使用旧版本的用户出错，因为他们将接收到他们不理解的格式的消息。
#log.message.format.version=0.10. [root@yinzhengjie ~]#
[root@yinzhengjie ~]# cat /soft/kafka/config/server.properties #上述配置文件详解
5>.修改kafka的启动脚本
[root@yinzhengjie ~]# cat /soft/kafka/bin/kafka-server-start.sh
#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License. if [ $# -lt ];
then
echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
exit
fi
base_dir=$(dirname $) if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
#在这里指定堆内存为20G
export KAFKA_HEAP_OPTS="-Xmx20G -Xms20G"
fi EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'} COMMAND=$
case $COMMAND in
-daemon)
EXTRA_ARGS="-daemon "$EXTRA_ARGS
shift
;;
*)
;;
esac exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
[root@yinzhengjie ~]#
6>.编写kafka启动脚本
[root@yinzhengjie ~]# cat /usr/local/bin/xkafka.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com #判断用户是否传参
if [ $# -ne ];then
echo "无效参数，用法为: $0 {start|stop}"
exit
fi #获取用户输入的命令
cmd=$ for (( i= ; i<= ; i++ )) ; do
tput setaf
echo ========== kafka${i}.aggrx $cmd ================
tput setaf
case $cmd in
start)
ssh kafka${i}.aggrx "source /etc/profile ; nohup kafka-server-start.sh /soft/kafka/config/server.properties >> /home/yinzhengjie/kafka/console/kafka-`date +%F`.log &"
#ssh kafka${i}.aggrx "source /etc/profile ; kafka-server-start.sh -daemon /soft/kafka/config/server.properties"
echo kafka${i}.aggrx "服务已启动"
;;
stop)
ssh kafka${i}.aggrx "source /etc/profile ; kafka-server-stop.sh"
echo kafka${i}.aggrx "服务已停止"
;;
*)
echo "无效参数，用法为: $0 {start|stop}"
exit
;;
esac
done
[root@yinzhengjie ~]#
7>.查看各个服务器的启动进程
[root@yinzhengjie ~]# cat /usr/local/bin/xcall.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com #判断用户是否传参
if [ $# -lt ];then
echo "请输入参数"
exit
fi #获取用户输入的命令
cmd=$@ #Kafka集群批量执行命令
for (( i=;i<=;i++ ))
do
#使终端变绿色
tput setaf
echo =========== kafka${i}.aggrx : $cmd ===========
#使终端变回原来的颜色，即白灰色
tput setaf
#远程执行命令
ssh kafka${i}.aggrx $cmd
#判断命令是否执行成功
if [ $? == ];then
echo "命令执行成功"
fi
done
[root@yinzhengjie ~]#
[root@yinzhengjie ~]# cat /usr/local/bin/xcall.sh #编写批量执行的脚本，需要你手动配置SSH免密要登录。
[root@yinzhengjie ~]# xcall.sh jps
=========== yinzhengjie.aggrx : jps ===========
Jps
Kafka
ProdServerStart
命令执行成功
=========== kafka117.aggrx : jps ===========
Jps
Kafka
QuorumPeerMain
命令执行成功
=========== kafka118.aggrx : jps ===========
QuorumPeerMain
Jps
Kafka
命令执行成功
=========== kafka119.aggrx : jps ===========
Jps
QuorumPeerMain
Kafka
命令执行成功
=========== kafka120.aggrx : jps ===========
Jps
Kafka
命令执行成功
[root@yinzhengjie ~]#