源起：工程現階段中間件采用的是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");