源起：工程现阶段中间件采用的是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 访问kafka与spring访问kafka的demo涉及安全信息就不给出了。