#!/bin/bash  
  
# Kafka 集群的服务器列表  
BOOTSTRAP_SERVERS="hadoop11:9092,hadoop12:9092,hadoop13:9092"  
  
# Kafka 脚本路径，根据你的实际安装路径修改  
KAFKA_SCRIPTS_DIR="/opt/installs/kafka/bin"  
  
# Kafka 消费者和生产者脚本的完整路径  
KAFKA_CONSUMER_SCRIPT="${KAFKA_SCRIPTS_DIR}/kafka-console-consumer.sh"  
KAFKA_PRODUCER_SCRIPT="${KAFKA_SCRIPTS_DIR}/kafka-console-producer.sh"  
  
# 检查是否提供了必要的参数  
if [ "$#" -lt 1 ]; then  
    echo "Usage: $0 <action> [<options>]"  
    echo "Action: create, list, describe, delete, consumer, producer start ,stop"  
    exit 1  
fi 
  
ACTION=$1  
TOPIC=$2  
  
# 根据传入的参数执行相应的操作  
case $ACTION in  
    create)  
        # 创建topic  
		echo "输入分区数:"  
        read PARTITIONS  
        echo "输入副本数:"  
        read REPLICATION_FACTOR  
        "${KAFKA_SCRIPTS_DIR}/kafka-topics.sh" --bootstrap-server $BOOTSTRAP_SERVERS --create --topic $TOPIC --partitions $PARTITIONS --replication-factor $REPLICATION_FACTOR  
		;;
	start)  
        # 开启kafka，在此处我的三个节点时hadoop11,hadoop12,hadoop13;可以按需修改


		for i in 11 12 13
		do
			if ssh hadoop$i "source /etc/profile; kafka-server-start.sh -daemon  /opt/installs/kafka/config/server.properties"; then
				echo "$hadoop$i:kafka开启成功"
			else
				echo"$hadoop$i:kafka开启失败"
			fi
		done
        ;;  	
	stop)  
        # 关闭kafka  
		for i in 11 12 13
		do
			if ssh hadoop$i "source /etc/profile; kafka-server-stop.sh stop"; then
				echo "$hadoop$i：关闭成功"
			else
				echo"$hadoop$i:kafka关闭失败"
			fi
		done
        ;; 
    list)  
        # 列出所有topic  
        "${KAFKA_SCRIPTS_DIR}/kafka-topics.sh" --bootstrap-server $BOOTSTRAP_SERVERS --list  
        ;;  
    describe)  
        # 查看topic详情  
        "${KAFKA_SCRIPTS_DIR}/kafka-topics.sh" --bootstrap-server $BOOTSTRAP_SERVERS --describe --topic $TOPIC  
        ;;  
    delete)  
        # 删除topic  
        # 注意：在某些Kafka版本中，需要设置`delete.topic.enable=true`在server.properties中  
        "${KAFKA_SCRIPTS_DIR}/kafka-topics.sh" --bootstrap-server $BOOTSTRAP_SERVERS --delete --topic $TOPIC  
        ;;  
    consumer)  
        # 运行Kafka消费者  
        "$KAFKA_CONSUMER_SCRIPT" --bootstrap-server $BOOTSTRAP_SERVERS --topic $TOPIC --from-beginning  
        ;;  
    producer)  
        # 运行Kafka生产者  
        "$KAFKA_PRODUCER_SCRIPT" --bootstrap-server $BOOTSTRAP_SERVERS --topic $TOPIC  
        ;;  
    *)  
        echo "Invalid action: $ACTION"  
        exit 1  
        ;;  
esac  
  
exit 0

使用说明：我的脚本名称是topics,并且将其放入/usr/local/bin/ 目录下,因此我的topics,相当于运行

sh topics.sh

1.选择想要进行的操作：create, list, describe, delete, consumer, producer start ,stop

 其中start,stop,list只需要传入相关操作

 create,  describe, delete, consumer, producer需要传入两个参数，<操作名称，队列名称>