中間件一鍵部署腳本：zookeeper/kafka集群模式

./deploy_kafka_cluster.sh 1.1.1.1 2.2.2.2 3.3.3.3

#!/bin/bash

kafka_tar_name=$(ls kafka*)

if [[ -z "${kafka_tar_name}" ]]; then
    echo "未找到 kafka 二進制安裝包"
    exit 1
fi

zookeeper_tar_name=$(ls *zookeeper*)

if [[ -z "${zookeeper_tar_name}" ]]; then
    echo "未找到 zookeeper 二進制安裝包"
    exit 1
fi

if [ $# -ne 3 ];then echo "ERROR: 腳本需要參數(shù)三個IP";exit ;fi

#解析參數(shù)
node1=$1
node2=$2
node3=$3

deployment_zookeeper(){
tar -xzf ${zookeeper_tar_name}
cd *zookeeper* 
zoo_currentdir=$(pwd)
mkdir data
#zookeeper配置
cat >conf/zoo.cfg<<EOF
tickTime=2000
initLimit=20
syncLimit=10
dataDir=$zoo_currentdir/data
dataLogDir=$zoo_currentdir/datalog
clientPort=2181
maxClientCnxns=2000
4lw.commands.whitelist=*
autopurge.snapRetainCount=3
autopurge.purgeInterval=1
server.1=$node1:2888:3888
server.2=$node2:2888:3888
server.3=$node3:2888:3888
EOF

cat >$zoo_currentdir/data/myid<<EOF
$1
EOF

#啟動zk
$zoo_currentdir/bin/zkServer.sh start
}

deployment_kafka(){
cd - >/dev/null
tar -xzf ${kafka_tar_name}
cd kafka* 
currentdir=$(pwd)

#kafka配置
cat >config/server.properties<<EOF
broker.id=$1
advertised.listeners=PLAINTEXT://$current_ip:9092
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=$currentdir/logs/kafka
num.partitions=3
num.recovery.threads.per.data.dir=1
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=$node1:2181,$node2:2181,$node3:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0
message.max.bytes=104857600
replica.fetch.max.bytes=104857600
EOF
cat >config/consumer.properties<<EOF
bootstrap.servers=localhost:9092
group.id=test-consumer-group
fetch.message.max.bytes=104857600
EOF


cat >config/producer.properties<<EOF
bootstrap.servers=localhost:9092
compression.type=none
max.request.size=104857600
acks=all
EOF

#啟動服務
$currentdir/bin/kafka-server-start.sh -daemon $currentdir/config/server.properties
}



#判斷當前服務器是否在節(jié)點中
##獲取當前節(jié)點的 IP（多個 IP 會用空格分隔）
current_ip_list=$(hostname -I)
current_ip=$(echo "$current_ip_list"|tr ' ' '\n' |grep -Fxf - <(echo "$node1 $node2 $node3"|tr ' ' '\n'))
if [ -z "$current_ip" ];then echo "ERROR: 當前服務器不在指定節(jié)點范圍中，退出部署";exit; fi

if [[ $current_ip == $node1 ]];then
echo "當前節(jié)點IP為node1:$current_ip"
deployment_zookeeper 1
deployment_kafka 1
elif [[ $current_ip == $node2 ]];then
echo "當前節(jié)點IP為node2:$current_ip"
deployment_zookeeper 2
deployment_kafka 2
else
echo "當前節(jié)點IP為node3:$current_ip"
deployment_zookeeper 3
deployment_kafka 3
fi


echo "zookeeper/Kafka 已啟動，請檢查 $current_ip:9092 端口是否可以正常訪問"