這是一個(gè)如何配置kafka支持SSL環(huán)境的例子。

我們的目標(biāo)：

1. 使用kafka2.1.1
2. 利用kafka自帶的zookeeper
3. 使用docker來啟動(dòng)kafka/zookeeper clustor

Step 1：生成jks文件

#!/bin/bash# generate CA certificate openssl req -new -x509 -keyout ca.key -out ca.pem -days 365 -passout pass:test1234 \ -subj "/C=cn/O=mycomp/OU=mygroup/CN=ca"# generate server keystore keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA \ -storepass test1234 -keypass test1234 -genkey \ -dname "C=cn,O=mycomp,OU=mygroup,CN=server" # generate client keystore keytool -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA \ -storepass test1234 -keypass test1234 -genkey \ -dname "C=cn,O=mycomp,OU=mygroup,CN=client"# import CA certificate into server truststore keytool -keystore server.truststore.jks -alias CARoot -import -noprompt \ -file ca.pem -storepass test1234 # import CA certificate into client truststore keytool -keystore client.truststore.jks -alias CARoot -import -noprompt \ -file ca.pem -storepass test1234# export server certificate keytool -keystore server.keystore.jks -alias localhost -certreq \ -file server.csr -storepass test1234 -ext san="DNS:kafka.example.com" # export client certificate keytool -keystore client.keystore.jks -alias localhost -certreq \ -file client.csr -storepass test1234 -ext san="DNS:kafka.example.com"# sign server certificate with CA certificate openssl x509 -req -CA ca.pem -CAkey ca.key -in server.csr \ -out server.pem -days 365 -CAcreateserial -passin pass:test1234 \ -extensions SAN \ -extfile <(cat /etc/pki/tls/openssl.cnf <(printf "[SAN]\nsubjectAltName=DNS:kafka.example.com")) # sign client certificate with CA certificate openssl x509 -req -CA ca.pem -CAkey ca.key -in client.csr \ -out client.pem -days 365 -CAcreateserial -passin pass:test1234 \ -extensions SAN \ -extfile <(cat /etc/pki/tls/openssl.cnf <(printf "[SAN]\nsubjectAltName=DNS:kafka.example.com"))# import CA certificate into server keystore keytool -keystore server.keystore.jks -alias CARoot -import -noprompt \ -file ca.pem -storepass test1234 # import CA certificate into client keystore keytool -keystore client.keystore.jks -alias CARoot -import -noprompt \ -file ca.pem -storepass test1234# import server certificate into server keystore keytool -keystore server.keystore.jks -alias localhost -import \ -file server.pem -storepass test1234 # import client certificate into client keystore keytool -keystore client.keystore.jks -alias localhost -import \ -file client.pem -storepass test1234 

運(yùn)行完之后，會(huì)生成server.keystore.jks/server.truststore.jks，client.keystore.jks/client.truststore.jks一共四個(gè)文件。

有一個(gè)地方要注意：
SAN的值給定義上了kafka.example.com，這兒實(shí)際上keytool是有bug的，實(shí)際應(yīng)用中，我們可能只能定義域名，無法定義確切的主機(jī)名，所以最好的方式是在這里定義DNS的值為".example.com"，但是由于keytool存在bug(設(shè)計(jì)問題)，他不支持"'開頭，那么我們只能使用全名主機(jī)名了。
要解決這個(gè)問題，可以使用openssl工具來生成"*.example.com"，不要直接使用keytool。

Step 2：配置zookeeper和kafka的配置文件

增加/修改zookeeper配置文件/opt/kafka/config/zookeeper.properties：

dataDir=/data clientPort=2181 maxClientCnxns=0 initLimit=5 dataLogDir=/datalog syncLimit=2 autopurge.purgeInterval=1 autopurge.snapRetainCount=3 server.1=zookeeper.example.com:2888:3888 tickTime=2000 

這個(gè)文件沒啥補(bǔ)充的。

增加/修改kafka配置文件/opt/kafka/config/server.properties文件：

broker.id=1 zookeeper.connect=zookeeper.example.com:2181 listeners=SSL://:9093 advertised.listeners=SSL://kafka.example.com:9093 listener.security.protocol.map=SSL:SSL # security.inter.broker.protocol=ssl ssl.client.auth=required ssl.keystore.location=/work/ssl/server.keystore.jks ssl.keystore.password=test1234 ssl.key.password=test1234 ssl.truststore.location=/work/ssl/server.truststore.jks ssl.truststore.type=JKS ssl.truststore.password=test1234 ssl.keystore.type=JKS 

這個(gè)文件有幾點(diǎn)說明：

1. ssl.client.auth是否驗(yàn)證client端的證書，即雙向驗(yàn)證。
2. listeners支持kafka同時(shí)支持SSL和PLAINTXT等多種模式
   listeners=PLAINTEXT://:9092,SSL://:9093
   這樣kafka就能支持在端口9092上監(jiān)聽非SSL的請求，同時(shí)在9093端口監(jiān)聽SSL端口。
3. security.inter.broker.protocol指定在kafka之間通信是是否使用SSL。
   在前面的listeners我們知道kafkakey監(jiān)聽多個(gè)端口，這樣運(yùn)行對client進(jìn)來的請求，我們讓他監(jiān)聽在9093的SSL端口，然后kafka之間的通信我們讓他監(jiān)聽的9092端口上，這樣保證了外網(wǎng)訪問的安全(SSL)，也提高了內(nèi)部通信的性能(PLAINTEXT)。
   這個(gè)內(nèi)部通信協(xié)議指得是kafka之間的通信，不包括kafka和zookeeper之間的通信，也就是說kafka和zookeeper之間的通信還是PLAINTEXT的。
4. advertised.listeners廣播監(jiān)聽地址
   前面的listeners可以配置0.0.0.0地址，但是advertised.listeners不能使用這個(gè)地址，因?yàn)檫@個(gè)地址會(huì)被SSL連接的時(shí)候驗(yàn)證SAN值；當(dāng)然也可以把listeners配置成SAN地址，這樣advertised.listeners就不需要了，但是這種配置，無法區(qū)分內(nèi)網(wǎng)訪問，還是外部訪問。

Step 3: 配置docker-compose.yaml

這里需要把kafka達(dá)到image里面進(jìn)去，構(gòu)造兩個(gè)images：kafka和zookeeper都使用kafka的安裝包就行。

version: '2' networks: byfn:services: zookeeper.example.com: image: zookeeper container_name: zookeeper.example.com environment: - ZOO_MY_ID=1 - ZOO_SERVERS=server.1=zookeeper.example.com:2888:3888 volumes: - ./zookeeper.properties:/opt/kafka/config/zookeeper.properties networks: - byfnkafka.example.com: image: kafka container_name: kafka.example.com environment: volumes: - ./ssl:/ssl - ./server.properties:/opt/kafka/config/server.properties depends_on: - zookeeper.example.com networks: - byfn 

Step 4: 啟動(dòng)clustor

$ docker-compose up 

Step 5: 客戶端訪問

配置client-ssl.properties文件

security.protocol=SSLssl.truststore.location=/ssl/client.truststore.jksssl.truststore.password=test1234 ssl.keystore.location=/ssl/client.keystore.jksssl.keystore.password=test1234 ssl.key.password=test1234 

啟動(dòng)producer和consumer：

# producer /opt/kafka/bin/kafka-console-producer.sh \ --broker-list kafka.example.com:9093 \ --topic test-topic \ --producer.config /ssl/client-ssl.properties# consumer /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server kafka1.example.com:9093 \ --topic test-topic \ --consumer.config /ssl/client-ssl.properties \ --from-beginning 

Step 6: 如果是雙端口配置

即9092是PLAINTEXT訪問， 9093是SSL訪問

# server.properties listeners=PLAINTEXT://:9092,SSL://:9093 advertised.listeners=PLAINTEXT://kafka.example.com:9092,SSL://kafka.example.com:9093 listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL 

那么client端既可以訪問9092，也可以訪問9093，以producer為例：

# PLAINTEXT /opt/kafka/bin/kafka-console-producer.sh \ --broker-list kafka.example.com:9092 \ --topic test-topic \# SSL /opt/kafka/bin/kafka-console-producer.sh \ --broker-list kafka.example.com:9093 \ --topic test-topic \ --producer.config /ssl/client-ssl.properties 

差別PLAINTEXT訪問9092端口，SSL訪問9093端口，并且配置上SSL證書信息。

總的來說，這個(gè)過程還是比較清楚的。

本頁內(nèi)容由塔燈網(wǎng)絡(luò)科技有限公司通過網(wǎng)絡(luò)收集編輯所得，所有資料僅供用戶學(xué)習(xí)參考，本站不擁有所有權(quán)，如您認(rèn)為本網(wǎng)頁中由涉嫌抄襲的內(nèi)容，請及時(shí)與我們聯(lián)系，并提供相關(guān)證據(jù)，工作人員會(huì)在5工作日內(nèi)聯(lián)系您，一經(jīng)查實(shí)，本站立刻刪除侵權(quán)內(nèi)容。本文鏈接:http://jstctz.cn/20390.html