1.概述 在Kafka0.9版本之前，Kafka叢集時沒有安全機制的。Kafka Client應用可以通過連線Zookeeper地址，例如zk1:2181:zk2:2181,zk3:2181等。來獲取儲存在Zookeeper中的Kafka元資料資訊。拿到Kafka Broker地址後，連線到Kafka叢集，就可以操作叢集上的所有主題了。由於沒有許可權控制，叢集核心的業務主題時存在風險的。

2.內容 2.2 身份認證

Kafka的認證範圍包含如下：

Client與Broker之間 Broker與Broker之間 Broker與Zookeeper之間 當前Kafka系統支援多種認證機制，如SSL、SASL（Kerberos、PLAIN、SCRAM）。

2.3 SSL認證流程

在Kafka系統中，SSL協議作為認證機制預設是禁止的，如果需要使用，可以手動啟動SSL機制。安裝和配置SSL協議的步驟，如下所示：

在每個Broker中Create一個Tmp金鑰庫 建立CA 給證書籤名 配置Server和Client 執行指令碼如下所示：

#! /bin/bash 2# 1.Create rsa 3keytool -keystore server.keystore.jks -alias dn1 -validity 365 -genkey -keyalg RSA 4# 2.Create CA 5openssl req -new -x509 -keyout ca-key -out ca-cert -days 365 6# 3.Import client 7keytool -keystore client.truststore.jks -alias CAROOT -import -file ca-cert 8# 4.Import server 9keytool -keystore server.truststore.jks -alias CAROOT -import -file ca-cert 10# 5.Export 11keytool -keystore server.keystore.jks -alias dn1 -certreq -file cert-file 12# 6.Signed 13openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:123456 14# 7.Import ca-cert 15keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert 16# 8.Import cert-signed 17keytool -keystore server.keystore.jks -alias dn1 -import -file cert-signed 2.4 SASL認證流程

在Kafka系統中，SASL機制包含三種，它們分別是Kerberos、PLAIN、SCRAM。以PLAIN認證為示例，下面給大家介紹PLAIN認證流程。

2.4.1 配置Server

首先，在$KAFKA_HOME/config目錄中新建一個檔案，名為kafka_server_jaas.conf，配置內容如下：

1KafkaServer { 2 org.apache.kafka.common.security.plain.PlainLoginModule required 3 username=“smartloli” 4 password=“smartloli-secret” 5 user_admin=“smartloli-secret”; 6}; 7 8Client { 9 org.apache.kafka.common.security.plain.PlainLoginModule required 10 username=“smartloli” 11 password=“smartloli-secret”; 12}; 然後在Kafka啟動指令碼（

1[[email protected] bin]$ vi kafka-server-start.sh 2 3# Add jaas file 4export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka/config/kafka_server_ 接下來，配置server.properties檔案，內容如下：

1# Set ip & port 2listeners=SASL_PLAINTEXT://dn1:9092 3advertised.listeners=SASL_PLAINTEXT://dn1:9092 4# Set protocol 5security.inter.broker.protocol=SASL_PLAINTEXT 6sasl.enabled.mechanisms=PLAIN 7sasl.mechanism.inter.broker.protocol=PLAIN 8 9# Add acl 10allow.everyone.if.no.acl.found=true 11auto.create.topics.enable=false 12delete.topic.enable=true 13advertised.host.name=dn1 14super.users=User:admin 15 16# Add class 17authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer 2.4.2 配置Client

當Kafka Server端配置啟用了SASL/PLAIN，那麼Client連線的時候需要配置認證資訊，Client配置一個kafka_client_jaas.conf檔案，內容如下：

1KafkaClient { 2 org.apache.kafka.common.security.plain.PlainLoginModule required 3 username=“admin” 4 password=“admin-secret”; 5}; 然後，在producer.properties和consumer.properties檔案中設定認證協議，內容如下：

1security.protocol=SASL_PLAINTEXT 2sasl.mechanism=PLAIN 最後，在kafka-console-producer.sh指令碼和kafka-console-producer.sh指令碼中新增JAAS檔案的路徑，內容如下：

1# For example: kafka-console-producer.sh [email protected] bin]$ vi kafka-console-producer.sh 3 4# Add jaas file 5export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka 6/config/kafka_client_jaas.conf"

2.5 ACL操作

在配置好SASL後，啟動Zookeeper叢集和Kafka叢集之後，就可以使用kafka-acls.sh指令碼來操作ACL機制。

（1）檢視：在kafka-acls.sh指令碼中傳入list引數來檢視ACL授權新

1[[email protected] bin]$ kafka-acls.sh --list --authorizer-properties zookeeper.connect=dn1:2181

（2）建立：建立待授權主題之前，在kafka-acls.sh指令碼中指定JAAS檔案路徑，然後在執行建立操作

1[[email protected] bin]$ kafka-topics.sh --create --zookeeper dn1:2181 --replication-factor 1 --partitions 1 --topic kafka_acl_topic （3）生產者授權：對生產者執行授權操作

1[[email protected] ~]$ kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=dn1:2181 --add --allow-principalUser:producer --operation Write --topic kafka_acl_topic （4）消費者授權：對生產者執行授權後，通過消費者來進行驗證

1[[email protected] ~]$ kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=dn1:2181 --add --allow-principalUser:consumer --operation Read --topic kafka_acl_topic （5）刪除：通過remove引數來回收相關許可權

1[[email protected] bin]$ kafka-acls.sh --authorizer-properties zookeeper.connect=dn1:2181 --remove --allow-principal User:producer --operation Write --topic kafka_acl_topic3 3.總結 在處理一些核心的業務資料時，Kafka的ACL機制還是非常重要的，對核心業務主題進行許可權管控，能夠避免不必要的風險。

歡迎工作一到五年的Java工程師朋友們加入Java架構開發： 854393687 群內提供免費的Java架構學習資料（裡面有高可用、高併發、高效能及分散式、Jvm效能調優、Spring原始碼，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構資料）合理利用自己每一分每一秒的時間來學習提升自己，不要再用"沒有時間“來掩飾自己思想上的懶惰！趁年輕，使勁拼，給未來的自己一個交代！