1、配置zookeeper，版本：3.6.3

• vi conf/zoo.cfg
  autopurge.purgeInterval=1
  authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider # 末尾添加

• vi conf/jaas_zk.conf 内容如下
  Server {
  org.apache.zookeeper.server.auth.DigestLoginModule required
  user_super="你的zookeeper密码"
  user_xiong="你的zookeeper密码";
  };

• Client {
  org.apache.zookeeper.server.auth.DigestLoginModule required
  username="xiong"
  password="你的zookeeper密码";
  };

• vi conf/java.env
  JVMFLAGS="-Djava.security.auth.login.config={你的zookeeper路径}/conf/jaas_zk.conf" # 添加jaas_zk.conf配置
  然后重启zookeeper

2、Kafka开启认证，版本2.12

1、 vi kafka/config/server.properties
新增下面配置，之前已有的配置项，需要把之前的相同的配置项删掉，比如之前已存在的 listeners 需要删掉， {ip_addr} 换成你的实际的ip

advertised.listeners=SASL_PLAINTEXT://{ip_addr}:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
listeners=SASL_PLAINTEXT://:9092

2、 vi kafka/config/kafka_server_jaas.conf
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="xiong"
password="你的kafka密码"
user_xiong="你的kafka密码";
};

ZkClient {
org.apache.zookeeper.server.auth.DigestLoginModule required
username="xiong"
password="你的zookeeper密码";
};

3、vi kafka/bin/kafka-run-class.sh
在 base_dir=0)/.. 后新增
export KAFKA_OPTS="-Dzookeeper.sasl.client=true -Dzookeeper.sasl.clientconfig=ZkClient -Dzookeeper.sasl.client.username=xiong -Djava.security.auth.login.config=/usr/local/xiong/kafka/config/kafka_server_jaas.conf"

4、vi kafka/config/producer.properties 新增下面配置
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \n username="xiong" \n password="你的kafka密码";

5、vi kafka/config/consumer.properties
新增下面配置
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \n username="xiong" \n password="你的kafka密码";

6、重启Kafka

7、：KAFKA_OPTS="-Dzookeeper.sasl.client=true -Dzookeeper.sasl.clientconfig=ZkClient -Dzookeeper.sasl.client.username=xiong -Djava.security.auth.login.config=/home/xiong/application/kafka_2.12-0.10.2.1/config/kafka_server_jaas.conf" /home/xiong/application/kafka_2.12-0.10.2.1/bin/zookeeper-security-migration.sh --zookeeper.acl secure --zookeeper.connect localhost:2181
效果：

image.png