环境部署

jdk1.8:https://download.oracle.com/otn/java/jdk/8u221-b11/230deb18db3e4014bb8e3e8324f81b43/jdk-8u221-linux-x64.tar.gz

kafka_2.13-3.5.1:

脚本

#!/bin/bash
##################### 准备工作 #####################
#1.服务器直接下载包可能会很慢，建议下载到本地后上传至服务器
#2.可自定义安装包存放目录
#3.注意修改IP地址

#安装包目录mydir，即执行脚本的当前目录
mydir=$(cd "$(dirname "$0")";pwd)
echo $mydir

#kafka&&jdk存放目录
kafka=/opt
jdk=/usr

#zookeeper&&kafka日志存放路径
mkdir /var/lagou/zookeeper/data -p
data=/var/lagou/zookeeper/data
log=/var/lagou/kafka/kafka-logs

#更改IP
ip1=172.16.20.25

#更改主机名
hostname1=node2
hostnamectl set-hostname $hostname1

#配置hosts文件
sed -i -e '$a # \n'$ip1' '$hostname1'' /etc/hosts

#更改brokerid=0.1.2
VAR='broker.id=0'

#zookeeper节点的标记1,2,3
MYID=1

#更改监控主机变量hostname1，hostname2，hostname3
listeners="$hostname1"

########################################################################################

#安装jdk-1.8
tar -zxvf $mydir/jdk-8u221-linux-x64.tar.gz -C $jdk
#修改/etc/profile&&重启配置
sed -i '$a # \nexport JAVA_HOME='$jdk'/jdk1.8.0_221 \nexport PATH=$PATH:$JAVA_HOME/bin' /etc/profile
source /etc/profile

#解压
tar -zxvf $mydir/kafka_2.13-3.5.1.tgz -C $kafka
#修改/etc/profile&&重启配置
sed -i '$a # \nexport KAFKA_HOME='$kafka'/kafka_2.13-3.5.1 \nexport PHTH=$PATH:$KAFKA_HOME/bin' /etc/profile
source /etc/profile
#修改server.properties文件
SERVER_CONF=$kafka/kafka_2.13-3.5.1/config/server.properties
sed -i -e '35a \listeners=PLAINTEXT://:9092' ${SERVER_CONF}
sed -i -e '39a \advertised.listeners=PLAINTEXT://'$listeners':9092' ${SERVER_CONF}
sed -i -e 's:log.dirs=.*:log.dirs='$log':g' ${SERVER_CONF}
sed -i -e 's|zookeeper.connect=.*|zookeeper.connect='$hostname1':2181/myKafka|g' ${SERVER_CONF}
sed -i '$a \listeners=SASL_PLAINTEXT://:9092 \nadvertised.listeners=SASL_PLAINTEXT://'$listeners':9092 \nsecurity.inter.broker.protocol=SASL_PLAINTEXT \nsasl.mechanism.inter.broker.protocol=PLAIN \nsasl.enabled.mechanisms=PLAIN \nauthorizer.class.name=kafka.security.authorizer.AclAuthorizer \nsuper.users=User:admin \nallow.everyone.if.no.acl.found=false' ${SERVER_CONF}

#部署zookeeper
sed -i -e 's:dataDir=.*:dataDir='$data':g' $kafka/kafka_2.13-3.5.1/config/zookeeper.properties
sed -i -e '$a \server.1='$hostname1':2881:3881' $kafka/kafka_2.13-3.5.1/config/zookeeper.properties
#节点标记
echo $MYID > $data/myid

#部署认证
#编辑kafka_server_jaas.conf
touch $kafka/kafka_2.13-3.5.1/config/kafka_server_jaas.conf
echo 'KafkaServer {' >> $kafka/kafka_2.13-3.5.1/config/kafka_server_jaas.conf
sed -i -e 'aorg.apache.kafka.common.security.plain.PlainLoginModule required \nusername="admin" \npassword="admin" \nuser_admin="admin" \nuser_kafka="kafka"; \n}; \nClient { \norg.apache.zookeeper.server.auth.DigestLoginModule required \nusername="kafka" \npassword="kafka"; \n};' $kafka/kafka_2.13-3.5.1/config/kafka_server_jaas.conf

#编辑kafka_client_jaas.conf
touch $kafka/kafka_2.13-3.5.1/config/kafka_client_jaas.conf
echo 'KafkaClient {' >> $kafka/kafka_2.13-3.5.1/config/kafka_client_jaas.conf
sed -i -e 'aorg.apache.kafka.common.security.plain.PlainLoginModule required \nusername="admin" \npassword="admin"; \n}; \nClient { \norg.apache.zookeeper.server.auth.DigestLoginModule required \nusername="admin" \npassword="admin"; \n};' $kafka/kafka_2.13-3.5.1/config/kafka_client_jaas.conf

#编辑zk_server_jaas.conf
touch $kafka/kafka_2.13-3.5.1/config/zk_server_jaas.conf
echo 'Server {' >> $kafka/kafka_2.13-3.5.1/config/zk_server_jaas.conf
sed -i -e '$a org.apache.zookeeper.server.auth.DigestLoginModule required \nusername="kafka" \npassword="kafka” \nuser_kafka="kafka"; \n};' $kafka/kafka_2.13-3.5.1/config/zk_server_jaas.conf
sed -i -e '29a \    export KAFKA_OPTS="-Djava.security.auth.login.config='$kafka'/kafka_2.13-3.5.1/config/zk_server_jaas.conf -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider -Dzookeeper.requireClientAuthScheme=sasl"' $kafka/kafka_2.13-3.5.1/bin/zookeeper-server-start.sh

#编辑内部客户端工具
sed -i -e '$a security.protocol=SASL_PLAINTEXT \nsasl.mechanism=PLAIN' $kafka/kafka_2.13-3.5.1/config/consumer.properties
sed -i -e '$a security.protocol=SASL_PLAINTEXT \nsasl.mechanism=PLAIN' $kafka/kafka_2.13-3.5.1/config/producer.properties

#修改Broker启动脚本
sed -i -e '30a \    export KAFKA_OPTS=" -Djava.security.auth.login.config='$kafka'/kafka_2.13-3.5.1/config/kafka_server_jaas.conf"' $kafka/kafka_2.13-3.5.1/bin/kafka-server-start.sh
sed -i -e '18a \    export KAFKA_OPTS=" -Djava.security.auth.login.config='$kafka'/kafka_2.13-3.5.1/config/kafka_client_jaas.conf"' $kafka/kafka_2.13-3.5.1/bin/kafka-console-producer.sh
sed -i -e '18a \    export KAFKA_OPTS=" -Djava.security.auth.login.config='$kafka'/kafka_2.13-3.5.1/config/kafka_client_jaas.conf"' $kafka/kafka_2.13-3.5.1/bin/kafka-console-consumer.sh

#Topic_JAAS 文件加载 SASL 凭证
touch $kafka/kafka_2.13-3.5.1/config/config.properties
echo 'sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";' >> $kafka/kafka_2.13-3.5.1/config/config.properties
sed -i -e '$a security.protocol=SASL_PLAINTEXT \nsasl.mechanism=PLAIN' $kafka/kafka_2.13-3.5.1/config/config.properties

#JMX监控
sed -i -e '214,220c JMX_PORT=9997 \nJMX_RMI_PORT=9996 \nISKAFKASERVER="false" \nif [[ "$*" =~ "kafka.Kafka" ]]; then \n        ISKAFKASERVER="true" \nfi \nif [  $JMX_PORT ] && [ "true" == "$ISKAFKASERVER" ]; then \n        KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.rmi.port=$JMX_RMI_PORT " \n        echo set KAFKA_JMX_PORT:$KAFKA_JMX_OPTS \nfi' $kafka/kafka_2.13-3.5.1/bin/kafka-run-class.sh

#启动zookeeper
$kafka/kafka_2.13-3.5.1/bin/zookeeper-server-start.sh -daemon $kafka/kafka_2.13-3.5.1/config/zookeeper.properties #后台运行

#启动kafka
$kafka/kafka_2.13-3.5.1/bin/kafka-server-start.sh -daemon $kafka/kafka_2.13-3.5.1/config/server.properties #后台运行

#(注意更换主路径)
#启动生产者
#/opt/kafka_2.13-3.5.1/bin/kafka-console-producer.sh --broker-list node2:9092 --topic test --producer.config /opt/kafka_2.13-3.5.1/config/producer.properties
#创建消费者
#/opt/kafka_2.13-3.5.1/bin/kafka-console-consumer.sh --bootstrap-server node2:9092 --topic test --from-beginning --consumer.config /opt/kafka_2.13-3.5.1/config/consumer.properties
#创建topic
#/opt/kafka_2.13-3.5.1/bin/kafka-topics.sh --create --bootstrap-server node2:9092 --partitions 2  --replication-factor 1 --topic test --command-config /opt/kafka_2.13-3.5.1/config/config.properties
#查看topic
#/opt/kafka_2.13-3.5.1/bin/kafka-topics.sh --list --bootstrap-server node2:9092 --command-config /opt/kafka_2.13-3.5.1/config/config.properties

生产消费&&topic创建

#(注意更换主路径)
#启动生产者
#/opt/kafka_2.13-3.5.1/bin/kafka-console-producer.sh --broker-list node2:9092 --topic test --producer.config /opt/kafka_2.13-3.5.1/config/producer.properties
#创建消费者
#/opt/kafka_2.13-3.5.1/bin/kafka-console-consumer.sh --bootstrap-server node2:9092 --topic test --from-beginning --consumer.config /opt/kafka_2.13-3.5.1/config/consumer.properties
#创建topic
#/opt/kafka_2.13-3.5.1/bin/kafka-topics.sh --create --bootstrap-server node2:9092 --partitions 2  --replication-factor 1 --topic test --command-config /opt/kafka_2.13-3.5.1/config/config.properties
#查看topic
#/opt/kafka_2.13-3.5.1/bin/kafka-topics.sh --list --bootstrap-server node2:9092 --command-config /opt/kafka_2.13-3.5.1/config/config.properties