源起：工程现阶段中间件采用的是kafka。满足了大数据的高吞吐，项目间的解耦合，也增强了工程的容错率与扩展性。但是在安全这一块还有漏洞，kafka集群中，只要网站内的任何人知道kafka集群的ip与topic,都可以肆无忌惮的往集群中的topic中发送数据与消费数据。
经过调研：kafka的sasl acl可以设置安全机制，给每个主题设置多个用户，不同的用户赋予不同的读写权限。
A B 俩个用户，A用户允许读写kafka中的topic1,B用户不允许读写kafka中的topic1,这就成功控制了kafka的读写权限。
于是开始了长期的探索与配置kafka和zookeeper的相关配置文件。在配置过程中踩了n多坑，终于成功搞定，满足了现下工程所需，带来了满满的成就感。
使用软件以及版本 jdk1.80_144、zookeeper3.4.10(3.5.7也可以)、kafka_2.13-2.5.0,切记kafka版本一定要使用正确，本人之前尝试了kafka2.1.0与2.1.2，按照kafka官网上的文档，书上的教程以及网上的教程进行配置后，均未生效。所以版本一定要选对 ！！！！！，若有其他版本将该功能实现的同学也可以留言一起交流。我在centos7.6 与ubuntun14.04、18.04进行过尝试且成功，该权限的设置应该与服务器无关。
选好版本后就可以在服务器上进行配置了，我是在单节点上配置成功后，再将该配置进行扩展到 zookeeper用3台服务器，kafka用3台服务器，且分别用java原生语言与springboot各自写了一份demo。可以成功的控制不同用户对不同topic的读写权限。
开始
使用命令 tar -zxvf 解压jdk zookeeper kakfa 服务器用户为root用户在/etc/profile文件内进行路径配置，普通用户在该用户的根目录节点中找到 .bashrc文件(该文件是隐藏的)进行路径配置。配置成功后输入jps,会显示一个jps进程如下图
然后开始zookeeper与kafka的配置
zookeeper的配置
1.进入到zookeeeper的conf目录 使用命令 cp zoo_sample.cfg zoo.cfg 复制一份zoo.cfg配置文件
2.使用vim zoo.cfg进入该配置文件
红框1是存放zookeeper的数据与日志的部分
红框2是zookeeper配置SASL支持，若是zookeeper集群的话，则authProvider.1后面的数字在不同的服务器上要不一样
红框3是配置zookeeper集群的，单节点不用考虑该配置。若是集群应添加 server.2 server.3...并且在红框1的dataDir目录下创建myid文件，且在文件内添加server.n中的数字n
3 在zookeeper的conf目录下添加 zoo_jaas.conf文件 ，添加账号认证信息
4 在zookeeper中添加kafka jar包的依
5 修改 zookeeper bin目录下的zkEnv.sh 脚本
vim zkEnv.sh打开该脚本，在最后添加红框内的内容。
6 启动zookeeper
zkServer.sh start ,若以上步骤都正确配置后，zookeeper会正常启动，运行jps命令，下图红框中的进程会成功启动。若不能成功启动，一定要检查上述步骤中的文件配置路径无误后，
再去日志目录中查看日志
kafka的配置
1.在kafka config目录中添加、修改 相应配置文件
红框1 是新建的文件。 kafka_server_jaas.conf是kafka服务器中需要的用户配置文件。kafka_client_jaas.conf与kafka_producer.jaas.conf是客户端的配置文件，在服务器添加这两个文件是方便在服务器内进行脚本测试，实际开发中客服端配置文件是要用java代码读取的，不需要服务器上的。
红框2是kafka自带的文件，这些文件需要修改。server.properties 是kafka的核心配置文件，我们会在里面配置相应的信息。consumer.properties与 producer.properties是消费者脚本与生产者脚本需要的配置文件，在服务器上用脚本测试需要这两个文件。实际开发中可以在java代码中配置相应内容，不需要服务器上的。各个文件的内容如下图所需
kafka_server_jaas.conf
参考网上大家都是这么配置的 KafkaServer中是 代表一些客户端用户，这些用户根据被赋予的权限来对kafka的主题进行相关操作。
Client中的用户要与zoo_jaas.cfg中的用户一致 user_producer="prod-sec"的意思是账户名为producer，密码是 prod-sec
kafka_client_jaas.conf kafka_producer_jaas.conf
这个文件里面放的是客户端用户了
server.properties
1是与zookeeper建立链接，我这个是用的本地的zookeeper,集群的同学另行配置。
2 便是给kafka配置SASL权限，其中zsh是超级用户，不受权限影响。
producer.proerties、consumer.properties内容添加如下
2 在kafka的bin目录下修改以下脚本信息
kafka-server-start.sh修改如下
kafka-topic.sh修改如下
kafka-console-producer.sh 与 kafka-console-consumer.sh修改如下
以上配置就全部完成，启动命令 kafka-server.sh ../config/server.propertie，kafka便可以成功启动，如果启动不成功，不要慌。多启动几次试试，就成功了。很奇怪，不知道为啥这样子。
3 使用 kafka-acl.sh脚本控制不同用户对指定topic的权限。
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic acl_0305 创建主题
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 -add --deny-principal User:* --operation Write --topic acl_0305 禁止所有主题对该用户的读写权限
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 -add --allow-principal User:consumer --operation Write --topic acl_0305 允许 consumer 用户读写该主题
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 -remove --deny-principal User:*--operation Write --topic acl_0305 取消所有用户对该主题的禁止
本人在测试过程中发现，这个版本的kafka acl脚本只能按照这样的顺序控制权限。先禁止所有用户，然后允许一个用户，再取消对所有用户的禁止。后面就可以正常的对每个用户再进行权限控制了(允许的用户可以访问主题，没有添加允许的用户不能访问主题)，严重怀疑kafka的acl源代码没有完善的控制权限的机制。而且每个用户对主题读写权限是一起控制的，读与写并没有分开。不管怎么样，这样已经满足当下的需求了，只不过操作麻烦，还有待完善。
JAVA代码
原生代码核心
1.在创建Properties(props)前读取客户端配置文件。2. 给props添加 sasl配置.3其他代码按照常见的kafka生产者发数据，消费者消费数据配置即可
System.setProperty("java.security.auth.login.config", "d:\\conf\\kafka_client_jaas.conf");
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");