Kafka集群创建

上次集群忘了写文档，这次集群创建zk和kafka放在了一起，版本和生产一致，所以使用低版本
2.8.6

一、准备配置

1.1、配置env

$ cat /etc/profile.d/kafka.sh
# Java Environment
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export PATH=$PATH:$JAVA_HOME/bin
# Kafka  Environment
export KAFKA_HOME=/data/kafka
export PATH=$PATH:$KAFKA_HOME/bin

$ cat /etc/profile.d/zookeeper.sh 
# ZooKeeper Environment
export ZOOKEEPER_HOME=/data/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin

1.2、启动文件

zk启动文件，forking是通过子进程去管理，适合守护进程

$ cat /etc/systemd/system/zookeeper.service
[Unit]
Description=zookeeper
After=syslog.target network.target[Service]
Type=forking
User=root
ExecStart=/data/zookeeper/bin/zkServer.sh start
ExecStop=/data/zookeeper/bin/zkServer.sh stop
Restart=on-failure
Restart=always
RestartSec=5[Install]
WantedBy=multi-user.target

kafka启动文件

$ cat /etc/systemd/system/kafka.service
[Unit]
Description=kafka
Requires=network.target remote-fs.target
After=network.target remote-fs.target[Service]
LimitNOFILE=infinity
LimitNPROC=infinity
Type=forking
ExecStart=/data/kafka/bin/kafka-server-start.sh -daemon /data/kafka/config/server.properties
ExecStop=/data/kafka/bin/kafka-server-stop.sh
Restart=on-failure[Install]
WantedBy=multi-user.target

1.3、服务配置文件

$ cat /data/zookeeper/conf/zoo.cfg 
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data
dataLogDir=/data/zookeeper/logs
clientPort=2181
server.15=10.198.170.15:2888:3888
server.16=10.198.170.16:2888:3888
server.17=10.198.170.17:2888:3888
4lw.commands.whitelist=*

kafka_75">二、kafka集群初始化

由于我们要启用broker之间的加密通信，所以需要主节点这里先生成管理员用户密码

# kafka的时候第一台节点先不使用加密运行，然后生成SECRM加密
$ cat /data/kafka/config/server.properties 
broker.id=1
delete.topic.enable=true
listeners=PLAINTEXT://10.198.170.15:9092
# 认证配置
#inter.broker.listener.name=SASL_PLAINTEXT
#listeners=SASL_PLAINTEXT://10.198.170.15:9092
#security.protocol=SASL_PLAINTEXT
#sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
#sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
# ACL配置
allow.everyone.if.no.acl.found=true
super.users=User:admin
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
num.network.threads=5
num.io.threads=8
socket.send.buffer.bytes=10240000
socket.receive.buffer.bytes=10240000
socket.request.max.bytes=1048576000
log.dirs=/data/kafka/data
num.partitions=1
default.replication.factor=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.198.170.15:2181,10.198.170.16:2181,10.198.170.17:2181
zookeeper.connection.timeout.ms=12000
zookeeper.session.timeout.ms=12000
group.initial.rebalance.delay.ms=500
log.flush.interval.messages=10000
log.flush.interval.ms=1000
num.replica.fetchers=3
replica.fetch.min.bytes=1
replica.fetch.max.bytes = 104857600
unclean.leader.election.enable=false
auto.create.topics.enable = true
min.isync.replicas=2
replica.socket.receive.buffer.bytes = 65536
replica.socket.timeout.ms = 30000
replica.lag.time.max.ms =5000
replica.fetch.wait.max.ms = 1000
log.message.timestamp.type=LogAppendTime
log.cleanup.policy = delete
log.roll.hours=168
broker.rack=kafka-rac1
message.max.bytes=10000000
request.timeout.ms=30000

启动kafka
systemctl start kafka

生成加密用户信息

$ kafka-configs.sh --bootstrap-server localhost:9092   --alter --add-config 'SCRAM-SHA-256=[password=gwz3CFEuec9cwzxnd]'   --entity-type users --entity-name admin
# 启动参数里面有这个文件
$ cat /data/kafka/config/kafka_server_scram_jaas.conf 
KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="gwz3CFEuec9cwzxnd";
};

生成后注释listeners=PLANTEXT...，取消注释认证配置下面的5行

listeners=PLAINTEXT://10.198.170.15:9092
# 认证配置
#inter.broker.listener.name=SASL_PLAINTEXT
#listeners=SASL_PLAINTEXT://10.198.170.15:9092
#security.protocol=SASL_PLAINTEXT
#sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
#sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN