1. 创建admin用户
如果选择SCRAM机制，首先需要创建用户凭证。假设要创建一个用户名为admin，密码为admin的用户。
# 进入Kafka的安装目录
cd /usr/local/kafka-3.7.0
bin/kafka-configs.sh --bootstrap-server hadoop01:9092 --alter --add-config 'SCRAM-SHA-256=[password=admin]' --entity-type users --entity-name admin
2. 创建JAAS配置文件
Kafka使用JAAS（Java Authentication and Authorization Service）文件进行安全配置。
创建一个JAAS配置文件（例如 /usr/local/kafka-3.7.0/config/kafka_server_jaas.conf），内容如下。这个文件包含了所有合法的用户及其密码。
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin";
};
注意：
(1). KafkaServer：这部分配置用于Broker进程自身的认证，以及验证来自客户端的连接。
(2). username和password是Broker之间通信的凭证，应与第一步创建的用户凭证一致。
(3). user_用户名="密码"的格式用于定义客户端用户。务必妥善保管此文件，并设置严格的文件权限。
(4). 如果Zookeeper也配置了SASL认证，你还需要在此文件中添加Client部分，用于Broker与Zookeeper之间的认证。
3. 修改Kafka Broker配置
cd /usr/local/kafka-3.7.0/config
编辑Kafka的 server.properties 配置文件，主要修改或添加以下参数:
# 监听器配置为SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://192.168.80.131:9092
advertised.listeners=SASL_PLAINTEXT://192.168.80.131:9092
# 指定Broker间通信使用的安全协议和SASL机制
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
# 启用SASL机制
sasl.enabled.mechanisms=SCRAM-SHA-256
# （可选但推荐）启用ACL授权，并设置超级用户
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin
4. 指定JAAS文件并启动Kafka Broker
（1）在启动Kafka时，需要通过环境变量指定JAAS配置文件的路径。
修改/etc/profile文件，将上面内容添加进去：
export KAFKA_OPTS="-Djava.security.auth.login.config=${KAFKA_HOME}/config/kafka_server_jaas.conf"
使环境变量生效
source /etc/profile
（2）启动Kafka
/usr/local/kafka-3.7.0/bin/kafka-server-start.sh -daemon /usr/local/kafka-3.7.0/config/server.properties
5. 配置SASL客户端
客户端（生产者和消费者）也需要进行相应配置才能连接上已启用SASL的Kafka。
在客户端的配置文件中指定以下参数：
vi kafka_client_jaas
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";
6. 使用admin用户查询所有topic
/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092 --command-config /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf --list
7. 使用admin用户删除topic
/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092 --command-config /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf --delete --topic zgy-tjg-fsy ;
8. 使用admin用户创建新用户iot
bin/kafka-configs.sh --bootstrap-server hadoop01:9092 --alter --add-config 'SCRAM-SHA-256=[password=Iot@123]' --entity-type users --entity-name iot –command-config config/kafka_client_jaas.conf
9. 使用iot用户查看topic
在客户端的配置文件中指定以下参数：
vi kafka_client_iot_jaas
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="iot" password="Iot@123";
使用iot用户查看topic，发现查看不到
/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092 --command-config /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf --list
10. 使用admin用户给iot设置ACL权限
# 首先授予主题读取权限
/usr/local/kafka-3.7.0/bin/kafka-acls.sh --bootstrap-server 192.168.80.131:9092 --add --allow-principal User:iot --operation Read --operation Describe --topic zgy-tjg-rcms --command-config /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf ;
# 然后授予消费组权限
/usr/local/kafka-3.7.0/bin/kafka-acls.sh --bootstrap-server 192.168.80.131:9092 --add --allow-principal User:iot --operation Read --operation Describe --group iot-group --command-config /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf ;
11. 使用iot用户查询所有topic
/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092 --command-config /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf --list
12. 使用iot用户进行消费
方式一： 使用配置文件（推荐）
在原来的命令中通过 --consumer.config 参数来指定这个配置文件。
/usr/local/kafka-3.7.0/bin/kafka-console-consumer.sh --bootstrap-server hadoop01:9092 --topic zgy-tjg-rcms --consumer.config /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf --group iot-group --from-beginning
方式二：通过命令行参数直接传递。
如果你只是想做一次快速测试，也可以将认证参数直接作为属性在命令中传递。
/usr/local/kafka-3.7.0/bin/kafka-console-consumer.sh --bootstrap-server hadoop01:9092 --topic zgy-tjg-rcms \
--consumer-property security.protocol=SASL_PLAINTEXT \
--consumer-property sasl.mechanism=SCRAM-SHA-256 \
--consumer-property "sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"iot\" password=\"Iot@123\";" \
--group iot-group \
--from-beginning
