一、概述

GSSAPI： 使用的Kerberos認證，可以集成目錄服務，比如AD。從Kafka0.9版本開始支持

?PLAIN： 使用簡單用戶名和密碼形式。從Kafka0.10版本開始支持

SCRAM： 主要解決PLAIN動態更新問題以及安全機制，從Kafka0.10.2開始支持

OAUTHBEARER： 基于OAuth 2認證框架，從Kafka2.0版本開始支持

二、配置SASL/PLAIN

????????

kafka配置文件server.properties配置

security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=PLAIN sasl.enabled.mechanisms=PLAIN authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer allow.everyone.if.no.acl.found=true auto.create.topics.enable=true listeners=SASL_PLAINTEXT://<IP Address>:9092 advertised.listeners=SASL_PLAINTEXT://<IP Address>:9092

? ? ??

2. 創建kafka_server_jaas.conf文件

????????這個文件是配置Broker服務端的JASS，在Kafka程序目錄的config目錄中創建kafka_servre_jass.conf

????????

KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret" user_admin="admin-secret"; }; Client { org.apache.zookeeper.server.auth.DigestLoginModule required username="admin" password="admin-secret"; };

????????username和password配置成具體的賬號和密碼

????????KafkaServer字段是用來配置broker間通信使用的用戶名和密碼以及客戶端連接時需要的用戶名和密碼，其中username和password是broker用于初始化連接到其他的broker，kafka用戶為broker間的通訊。在一個Kafka集群中，這個文件的內容要一樣，集群中每個Borker去連接其他Broker的時候都使用這個文件中定義的username和password來讓對方進行認證。

????????Client部分是用來設置與Zookeeper的連接的，它還允許broker設置 SASL ACL 到zookeeper 節點，鎖定這些節點，只有broker可以修改它。如果Zookeeper與Broker之間不設置認證，那么就可以不配置Client部分。

?3.?啟動時指定安全配置

export KAFKA_OPTS="-Djava.security.auth.login.config=/KAFKA_HOME/config/kafka_server_jaas.conf"

或將以上指令寫入到啟動腳本中bin/kafka-server-start.sh

4.??啟動kafka server

bin/kafka-server-start.sh -daemon config/server.properties

5.?kafka client認證

? ?(1) 新建kafka_client_jaas.conf文件

? ? ?

KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret"; }; Client {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="admin"password="admin-secret"; };

username和password是真實配置的用戶名密碼

(2)?使用kafka自帶腳本消費消息

$ export KAFKA_OPTS="-Djava.security.auth.login.config=/KAFKA_HOME/config/kafka_client_jaas.conf"$ ./bin/kafka-console-consumer.sh --topic test-topic --from-beginning ?--consumer.config=config/consumer.properties?--bootstrap-server=localhost:9092

(3)??使用kafka自帶腳本生產消息

$ export KAFKA_OPTS="- Djava.security.auth.login.config=/KAFKA_HOME/config/kafka_client_jaas.conf" $ ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic --producer.config=config/producer.properties

6. Java程序使用

引入依賴

<!-- kafka-clients --> <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version> </dependency>

kafka-clients 3.0版本

配置

security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret"; Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //配置security.protocol properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, config.getSecurity().getProtocol()); //配置sasl.mechanism properties.put(SaslConfigs.SASL_MECHANISM, config.getSasl().getMechanism()); //配置sasl.jaas.config properties.put(SaslConfigs.SASL_JAAS_CONFIG, config.getSasl().getJaasConfig());KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

感謝：

https://www.cnblogs.com/rexcheny/articles/12884990.html https://docs.vmware.com/en/VMware-Smart-Assurance/10.1.0/sa-ui-installation-config-guide-10.1.0/GUID-3E473EC3-732A-4963-81BD-13BCCD3AC700.html

總結

以上是生活随笔為你收集整理的Kafka安全认证（Java)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。