kafka操作脚本

ゝ一世哀愁。 2022-05-27 02:29 139阅读 0赞

#!/bin/bash
    source /etc/profile
    if [ $1 = "start" ]
    then
        echo "------------ensure that zookeeper is started------------------"
        cat /data/kafka_2.11-0.10.1.0/config/nodes | while read host
        do
        {
            echo "$host is starting..."
            ssh $host "source /etc/profile;nohup /data/kafka_2.11-0.10.1.0/bin/kafka-server-start.sh /data/kafka_2.11-0.10.1.0/config/server.properties >> /home/elk/kafka.log 2>&1 &"
        }&
        wait
        done    
    elif [ $1 = "stop" ]
    then
        cat /data/kafka_2.11-0.10.1.0/config/nodes | while read host
        do
        {
            echo "$host is stoping..."
            ssh $host "source /etc/profile;nohup /data/kafka_2.11-0.10.1.0/bin/kafka-server-stop.sh /data/kafka_2.11-0.10.1.0/config/server.properties >> /home/elk/kafka.log 2>&1 &"
        }&
        wait
        done
    elif [ $1 = "producer" ]
    then
        kafka-console-producer.sh --broker-list localhost:9092 --topic $2 # 启动生产者，$2是topic
    elif [ $1 = "consumer" ]
    then
        kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic $2 #启动消费topic
    elif [ $1 = "delete" ]
    then
        kafka-topics.sh --delete --zookeeper localhost:2181 --topic $2 #删除topic
    elif [ $1 = "create" ]
    then
        kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 3 --topic $2 #创建topic
    elif [ $1 = "list" ]
    then
        kafka-topics.sh --list --zookeeper localhost:2181 #查看所有topic
    elif [ $1 = "desc" ]
    then
    	kafka-topics.sh --describe --zookeeper localhost:2181 --topic $2 #查看topic详细信息
    elif [ $1 = "assign" ]
    then	
    	kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file $2 --execute #执行分区分配
    elif [ $1 = "verify" ]
    then	
    	kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file $2 --verify #检查分区分配结果
    else
        echo "parameter invalid"
    fi