簡介

kafka 0.8.2以前官方沒有提供對安全機制的支援，在github上，有個叫kafka-ssl的開源專案，本文將介紹該專案的安裝與使用。

安裝

下載kafka-ssl的最新版本 自定義服務端和客戶端的keystore認證檔案 生成server的keystorekeytool -genkey -alias server -keypass kafkasure -keyalg RSA -keysize 1024-validity 365-keystore myserver.keystore -storepass surekafka -dname "CN=opensure, OU=xxxx, O=xxxx

, L=shanghai, ST=shanghai, C=CN"匯出server的trustedCertEntrykeytool -v -export -file myserver.cer -keystore myserver.keystore -alias server -storepass surekafka生成client的keystorekeytool -genkey -alias client -keypass kafkasure -keyalg RSA -keysize 1024-validity 365-keystore myclient.keystore -storepass surekafka

-dname "CN=opensure, OU=xxxx, O=xxxx, L=shanghai, ST=shanghai, C=CN"匯出client的trustedCertEntrykeytool -v -export -file myclient.cer -keystore myclient.keystore -alias client -storepass surekafka將server的trustedCertEntry匯入到client的keystore中keytool -import -keystore myclient.keystore -file myserver.cer -

alias server -keypass kafkasure -storepass surekafka輸入：是將client的trustedCertEntry匯入到server的keystore中keytool -import -keystore myserver.keystore -file myclient.cer -alias client -keypass kafkasure -storepass surekafka輸入：是這時檢視client和server的keystore檔案就都有PrivateKeyEntry和trustedCertEntry了keytool -list -v -keystore myserver.keystore -storepass surekafkakeytool -list -v -keystore myclient.keystore -storepass surekafka client.security.properties
# Keystore filekeystore.type=jkskeystore=config/myclient.keystorekeystorePwd=surekafkakeyPwd=kafkasure# Truststore filetruststore=config/myclient.keystoretruststorePwd=surekafka server.security.properties
#type of keystorekeystore.type=jks# Request client authwant.client.auth=true# Require client authneed.client.auth=true# Keystore filekeystore=config/myserver.keystorekeystorePwd=surekafkakeyPwd=kafkasure# Truststore filetruststore=config/myserver.keystoretruststorePwd=surekafka #打包成release的gz檔案./gradlew releaseTarGz#啟動zookeeperbin/zookeeper-server-start.sh config/zookeeper.properties#啟動kafka brokerbin/kafka-server-start.sh config/server.properties#建立topicbin/kafka-topics.sh --create --zookeeper name:2182--replication-factor 1--partitions 1--topic test2#生產bin/kafka-console-producer.sh  --broker-list name:9092--secure --client.security.file config/client.security.properties --topic test2#消費bin/kafka-console-consumer.sh --security.config.file config/client.security.properties -zookeeper name:2182--topic test2 #打包成release的gz檔案./gradlew releaseTarGz#啟動zookeeperbin/zookeeper-server-start.sh config/zookeeper.properties#啟動kafka brokerbin/kafka-server-start.sh config/server.properties#建立topicbin/kafka-topics.sh --create --zookeeper name:2182--replication-factor 1--partitions 1--topic test2#生產bin/kafka-console-producer.sh  --broker-list name:9092--secure --client.security.file config/client.security.properties --topic test2#消費bin/kafka-console-consumer.sh --security.config.file config/client.security.properties -zookeeper name:2182--topic test2

使用

Java Api消費示例：kafka-ssl0.8.2\src\consumer\ConsumerGroupExample.java Java Api生產示例：kafka-ssl0.8.2\src\consumer\ReadLocalProducer.java
注意： 1、注意在windows上執行會沒有反映和輸出，在linux上執行即可，依賴的lib可以在release的lib目錄下找到。 2、java api中消費時需指定引數： props.put("security.config.file","/opt/kafka_2.10-0.8.2-SNAPSHOT/config/client.security.properties"); 3、java api中生產是需指定引數： props.put("secure","true");props.put("client.security.file","/opt/kafka_2.10-0.8.2-SNAPSHOT/config/client.security.properties"); #打包成release的gz檔案./gradlew releaseTarGz#啟動zookeeperbin/zookeeper-server-start.sh config/zookeeper.properties#啟動kafka brokerbin/kafka-server-start.sh config/server.properties#建立topicbin/kafka-topics.sh --create --zookeeper name:2182--replication-factor 1--partitions 1--topic test2#生產bin/kafka-console-producer.sh  --broker-list name:9092--secure --client.security.file config/client.security.properties --topic test2#消費bin/kafka-console-consumer.sh --security.config.file config/client.security.properties -zookeeper name:2182--topic test2