kafka集群安装

这个是在 单机版安装后的前提下 部署的集群，

这样操作比较方便。

（1） 先安装单机版 kafka

kafka安装（单机版）： 连接

（2）同步kafka程序包到其他节点

data_rsync.sh /baimei/softwares/kafka_2.12-3.3.2

data_rsync.sh /etc/profile.d/kafka.sh

(3)修改各节点的配置文件

(1)broker.id

for ((host_id=111;host_id<=113;host_id++)) do ssh elk${host_id} "sed -ri 's#(broker.id=)111#\1${host_id}#g' $KAFKA_HOME/config/server.properties";done

检查一下：

egrep -v "^#|^$" $KAFKA_HOME/config/server.properties

（2）listeners

for ((host_id=111;host_id<=113;host_id++)) do ssh elk${host_id} "sed -ri 's#(listeners=PLAINTEXT://10.0.0.)111(:9092)#\1${host_id}\2#g' $KAFKA_HOME/config/server.properties";done

listeners 套接字服务器监听的地址。如果没有配置，主机名将等于的值

egrep -v  '^$|^#' $KAFKA_HOME/config/server.properties

（4）各节点启动服务

for ((host_id=111;host_id<=113;host_id++)) do ssh elk${host_id} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties ";done

kafka 集群 启动脚本：

自己编写：manager_kafka.sh

[root@baimeidashu-elk111 /usr/local/sbin]#cat  /usr/local/sbin/manager_kafka.sh
#!/bin/bash

#判断用户是否传参
if [ $# -ne 1 ];then
    echo "无效参数，用法为: $0  {start|stop|restart|status}"
    exit
fi

#获取用户输入的命令
cmd=$1

#定义函数功能
function kafkaManger(){
    case $cmd in
    start)
        echo "启动服务"        
        remoteExecution start
        ;;
    stop)
        echo "停止服务"
        stopKafka stop
        ;;
    restart)
        echo "重启服务"
        remoteExecution restart
        ;;
    status)
        echo "查看状态"
        remoteExecution status
        ;;
    *)
        echo "无效参数，用法为: $0  {start|stop|restart|status}"
        ;;
    esac
}


#定义执行的命令
function remoteExecution(){
    for (( i=111 ; i<=113 ; i++ )) ; do
            tput setaf 2
            echo ========== elk${i} kafaka-server-start.sh   $1 ================
            tput setaf 9
            ssh elk${i} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties" 
    done
}

function stopKafka(){
    for (( i=111 ; i<=113 ; i++ )) ; do
            tput setaf 2
            echo ========== elk${i} kafaka-server-stop.sh   $1 ================
            tput setaf 9
            ssh elk${i} "kafka-server-stop.sh -daemon $KAFKA_HOME/config/server.properties" 
    done

}


#调用函数
kafkaManger

记得修改 执行权限

启动：

manager_kafka.sh start

检查；

jps

每个节点有 kafka 进程就说明成功了。

（5） 创建 主题测试

创建一个名为"baimei-dashu87"的topic，分区数为3。

kafka-topics.sh --bootstrap-server 10.0.0.111:9092,10.0.0.112:9092 --create --topic baimei-dashu87 --partitions 3