Kafka-批量启动和停止Kafka节点脚本

1、编辑脚本

#!/bin/bash

zookeeper_home=/usr/local/zookeeper
kafka_home=/usr/local/kafka

zookeeper_array=(hadoop01 hadoop02 hadoop03)
kafka_array=(hadoop01 hadoop02 hadoop03)

## 启动和停止Zookeeper
function zookeeper_operate(){
    zookeeper_operate=$@
    if [[ ${zookeeper_operate} == "status" ]]
    then
        echo "~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"
        echo "About to check  zookeeper  status ...."
        sleep  5
    fi
    echo "****************************************Zookeeper ${zookeeper_operate}*******************************************"
    for zookeeper_node in ${zookeeper_array[@]}
    do
        echo "======================${zookeeper_node} ${zookeeper_operate}========================"
        ssh ${zookeeper_node} "source /etc/profile;${zookeeper_home}/bin/zkServer.sh  ${zookeeper_operate}"
        if [[ ${zookeeper_node} == "stop" ]]
        then
            ssh  ${zookeeper_node}  "rm -rf ${zookeeper}/logs/* "
        fi
    done

}

## 启动和停止Kafka
function kafka_operate(){
        kafka_operate=$@
        echo "****************************************Kafka ${kafka_operate}*******************************************"
        for kafka_node in ${kafka_array[@]}
        do
        echo "======================${kafka_node} ${kafka_operate}========================"
                case ${kafka_operate} in 
            start)
                echo ${kafka_home}/bin/kafka-server-start.sh -daemon  ${kafka_home}/config/server.properties
                ssh ${kafka_node} "source /etc/profile;${kafka_home}/bin/kafka-server-start.sh -daemon  ${kafka_home}/config/server.properties"
            ;;
            stop)
                echo ${kafka_home}/bin/kafka-server-stop.sh
                ssh ${kafka_node} "source /etc/profile;${kafka_home}/bin/kafka-server-stop.sh"
                ## 删除Kafka日志
                ssh ${kafka_node} "rm -rf  ${kafka_home}/logs/* "
            ;;
            *)
                echo "Input wrong,please check!"
                break
        esac 
                    
        done
    ## 等待5秒，确保zookeeper启动
    sleep 5

}

## 检查所有节点的状态
function jps_all_node(){
    check_str=$@
        echo "****************************************Jps check*******************************************"
        for jps_node in ${kafka_array[@]}
        do
        sleep 5
        echo "======================${jps_node} jps result========================" 
        ssh ${jps_node} "source /etc/profile ; jps | grep ${check_str}"
        done
}


echo  "=======Start Kafka cluster,please intput : start or 1 ======"
echo  "=======Stop  Kafka cluster,please intput : stop  or 0 ======"
read -p "please input : "  inputStr
case ${inputStr} in 
    start|START|1)
        ## 启动zookeeper 
        zookeeper_operate  start 
        ## 启动kafka
        kafka_operate   start
        ## 检查所有节点的状态
        jps_all_node  Kafka
    ;;
    stop|STOP|0)
        kafka_operate   stop
        ## 检查所有节点的状态
        jps_all_node  Kafka
    ;;
    *)
        echo "Input wrong,please check!"
        break
esac

2、启动Kafka

 sh manage-kafka.sh

 输入start或者1，即可启动所有的Kafka节点。

在Kafka中配置了zookeeper，所以脚本中会自动启动zookeeper服务。

3、停止kafka

 sh manage-kafka.sh

  输入stop或者0，即可停止所有的Kafka节点。