环境centos7
kafka集群和zookeeper集群默认都是不带用户密码的。

1. 配置zookeeper集群SASL

zookeeper所有节点都是对等的，只是各个节点角色可能不相同。以下步骤所有的节点配置相同。

1，为zookeeper添加SASL支持，在配置文件zoo.cfg添加如下配置

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

2，新建zoo_jaas.conf文件，为Zookeeper添加账号认证信息
这个文件你放在哪里随意，只要后面zkEnv配置正确的路径就好了。我是放在zook的conf路径下。zk_server_jaas.conf文件的内容如下

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="baoxue.123!"
    user_admin="baoxue.123!";
};

注解:
username和paasword是zk集群之间的认证密码。
user_admin=“baoxue.123!“定义了一个用户"admin”，密码是"baoxue.123!”,是kafka连接zk集群时用的
3.将Kafka相关jar包导入到Zookeeper
Zookeeper的认证机制是使用插件，

org.apache.kafka.common.security.plain.PlainLoginModule

所以需要导入Kafka相关jar包，kafka-clients相关jar包，在部署kafka服务下的libs目录中可以找到，根据kafka不同版本，相关jar包版本会有所变化。
所需要jar包如下，在zookeeper下创建目录zk_sasl_lib将jar包放入（目录名与位置可以随便，后续引用指定即可）：

kafka-clients-1.1.1.jar lz4-java-1.4.1.jar slf4j-api-1.7.25.jar
slf4j-log4j12-1.7.25.jar snappy-java-1.1.7.1.jar
4，修改zkEnv.sh,主要目的就是将这几个jar包使zookeeper读取到
在$ZK_HOME/bin目录下找到zkEnv.sh文件，添加如下代码,注意引用的目录下jar包，与之前创建的zoo_jaas.conf文件

for i in /home/zookeeper/zookeeper/zk_sasl_lib/*.jar;
do
    CLASSPATH="$i:$CLASSPATH"
done
export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/home/zookeeper/
zookeeper/conf/zk_server_jaas.conf"

5,重启zook集群服务
等所有节点修改完成后，规范重启zook集群，停从停主，起主起从
查看日志看是否有报错

2. 配置kafka集群sasl

所有节点操作相同

1，创建kafka_server_jaas.conf文件，该文件名可以自己修改，为kafka添加认证信息

注：内容如下（这里的Client与Zookeeper相对应，KafkaServer与后期调用时读取的KafkaClient相对应，是消费生产的账号密码，不要弄混了）：

KafkaServer {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="baoxue.123!"
 user_admin="baoxue.123!";
};

Client{
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="baoxue.123!";
};

解释:
KafkaServer中，使用user_来定义多个用户，供客户端程序（生产者、消费者程序）认 证 使用，可以定义多个，后续配置可能还可以根据不同的用户定义ACL，username=“admin”
password=“baoxue.123!” 这两个是kafka集群内部认证时用到的。
Client中，主要是broker链接到zookeeper，需要和之前配置的zk_server_jaas.conf里的 user_admin=“baoxue.123!”;对应，如果zk文件中配置了多个，填写其中一个即可。

2，在Kafka Server.properties添加、修改如下信息

listeners=SASL_PLAINTEXT://192.168.2.xxx:19092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

3.Kafka启动脚本中加入配置，读取第一步创建的文件,kafka_server_jaas.conf
修改kafka的kafka-server-start.sh文件，
找到如下代码

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"

修改成如下模样

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.
config=/opt/kafka_2.11-1.1.1/config/kafka_server_jaas.conf"

4.启动Kafka服务，查看日志是否正常，
所有节点修改完毕后，规范重启kafka集群，查看日志是否成功
至此kafka安全认证完成，
kafka_server_jaas.conf文件里KafkaServer部分的user_admin=“baoxue.123!”;里的用户和密码就是客户端程序连接kafka集群时的用户密码