环境介绍
节点说明
kafka1: 192.168.174.100（内） 192.168.174.200（外）
kafka2: 192.168.174.101 192.168.174.201
kafka3: 192.168.174.102 192.168.174.202
安装 java
wget https://download.java.net/java/GA/jdk25/bd75d5f9689641da8e1daabeccb5528b/36/GPL/openjdk-25_linux-x64_bin.tar.gz
cat > /etc/profile.d/jdk.sh << EOF
export JAVA_HOME=/usr/local/jdkexport PATH=$JAVA_HOME/bin:$JAVA_HOME/jre/bin:$PATHexport CLASSPATH=.$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib:$JAVA_HOME/lib/tools.jar
EOF
创建 kafka 用户
# groupadd kafka && useradd -M -N -g kafka -s /bin/false -c "kafka Server" kafka
下载 kafka
# wget https://dlcdn.apache.org/kafka/4.2.0/kafka_2.13-4.2.0.tgz
解压 kafka
# tar xf kafka_2.13-4.2.0.tgz -C /data/kafka
# ln -sv /data/kafka/kafka_2.13-4.2.0 /data/kafka/kafka
配置 kafka
配置文件路径：config/server.properties
process.roles=broker,controller
node.id=1
controller.quorum.bootstrap.servers=192.168.174.100:19093,192.168.174.101:19093,192.168.174.102:19093
listeners=PLAINTEXT://192.168.174.100:54111,CONTROLLER://192.168.174.100:19093,EXTERNAL://192.168.174.200:54910
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT
num.network.threads=32
num.io.threads=64
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=8388608
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka/data
num.partitions=18
num.recovery.threads.per.data.dir=4
offsets.topic.replication.factor=3
share.coordinator.state.topic.replication.factor=3
share.coordinator.state.topic.min.isr=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
log.retention.hours=8
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=0
message.max.bytes=5242880
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
allow.everyone.if.no.acl.found=false
super.users=User:admin;User:ANONYMOUS;User:test
部署 kafka
生成 UUID
CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_0_UUID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_1_UUID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_2_UUID="$(bin/kafka-storage.sh random-uuid)"
设置日志目录的格式
bin/kafka-storage.sh format --cluster-id ${CLUSTER_ID} \--initial-controllers "1@192.168.174.100:19093:${CONTROLLER_0_UUID},2@192.168.174.101:19093:${CONTROLLER_1_UUID},3@192.168.174.102:19093:${CONTROLLER_2_UUID}" \--config config/server.properties
Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=29) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=29, source=format command)
Formatting dynamic metadata directory /data/kafka/kafka/data with metadata.version 4.2-IV1.
firewalld 放行端口
firewall-cmd --permanent --zone=public --add-rich-rule='rule family="ipv4" source address="192.168.174.100" port protocol="tcp" port="54111" accept'
firewall-cmd --permanent --zone=public --add-rich-rule='rule family="ipv4" source address="192.168.174.100" port protocol="tcp" port="19093" accept'
firewall-cmd --reload
systemctl 启动 kafka
cat > /lib/systemd/system/kafka.service << EOF
[Unit]
Description=Apache Kafka
Documentation=http://kafka.apache.org/
After=network.target [Service]
Type=simple
User=kafka
Environment="JAVA_HOME=/usr/local/jdk"
Environment="PATH=\$JAVA_HOME/bin:\$PATH"
ExecStart=/data/kafka/kafka/bin/kafka-server-start.sh /data/kafka/kafka/config/server.properties
ExecStop=/data/kafka/kafka/bin/kafka-server-stop.sh
LimitNOFILE=1000000
TimeoutStopSec=180
Restart=on-failure[Install]
WantedBy=multi-user.target
EOF
认证配置
kafka_server_jaas.conf
KafkaServer{org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="xxxx";
};KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="xxxx";
}
更新 bin/kafka-server-start.sh
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/data/kafka/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"
动态创建账号
bin/kafka-configs.sh --bootstrap-server 192.168.174.100:54111,192.168.174.101:54111,192.168.174.102:54111 \--alter \--add-config 'SCRAM-SHA-256=[iterations=8192,password=xxxxxx]' \--entity-type users \--entity-name test
授权 topic 读写
bin/kafka-acls.sh --bootstrap-server 192.168.174.100:54111,192.168.174.101:54111,192.168.174.102:54111 --add --allow-principal User:test --operation Read --operation Write --topic test
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test, patternType=LITERAL)`: (principal=User:test, host=*, operation=READ, permissionType=ALLOW)(principal=User:test, host=*, operation=WRITE, permissionType=ALLOW)
授权消费组读
bin/kafka-acls.sh --bootstrap-server 192.168.174.100:54111,192.168.174.101:54111,192.168.174.102:54111 --add --allow-principal User:test --operation Read --group "*" --topic test
Adding ACLs for resource `ResourcePattern(resourceType=GROUP, name=*, patternType=LITERAL)`: (principal=User:test, host=*, operation=READ, permissionType=ALLOW)
查看授权列表
bin/kafka-acls.sh --bootstrap-server 192.168.174.100:54111,192.168.174.101:54111,192.168.174.102:54111 --list
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test, patternType=LITERAL)`: (principal=User:test, host=*, operation=READ, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=*, patternType=LITERAL)`: (principal=User:test, host=*, operation=READ, permissionType=ALLOW)
删除账号
bin/kafka-configs.sh --bootstrap-server 192.168.174.100:54111,192.168.174.101:54111,192.168.174.102:54111 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name test
测试 kafka
client.properties
# SASL 认证配置（PLAIN 机制示例）
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
#sasl.mechanism=PLAIN
#sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \username="tyyyanyiJYW-test" \password="YanYiJyw123qaz~!@";# 可选：增加超时时间
request.timeout.ms=60000
创建 topic
# bin/kafka-topics.sh --create --topic wgs-test-event --bootstrap-server 192.168.174.100:54111 --partitions 3 --replication-factor 2
Created topic wgs-test-event.
查看 topic
# bin/kafka-topics.sh --list --bootstrap-server 192.168.174.100:54111
__consumer_offsets
wgs-test-event
产生消息
# bin/kafka-console-producer.sh --topic wgs-test-event --bootstrap-server 192.168.174.200:54110 --command-config client.properties
消费消息
# bin/kafka-console-consumer.sh --topic wgs-test-event --from-beginning --bootstrap-server 192.168.174.100:54111 --command-config client.properties
删除 topic
# bin/kafka-topics.sh --delete --topic wgs-test-event --bootstrap-server 192.168.174.100:54111
