1. 准备证书文件

mkdir /opt/kafka/pkicd !$# 生成CA证书
openssl req -x509 -nodes -days 3650 -newkey rsa:4096 -keyout ca.key -out ca.crt -subj "/CN=Kafka-CA"# 生成私钥
openssl genrsa -out kafka.key 4096# 生成证书签名请求 (CSR)
openssl req -new -key kafka.key -out kafka.csr -subj "/CN=kafka-cluster"# 创建包含所有节点的SAN 配置文件
cat > san.cnf << EOF
[ req ]
distinguished_name = req_distinguished_name
req_extensions = req_ext
prompt = no[ req_distinguished_name ]
CN = kafka-cluster[ req_ext ]
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth, clientAuth
subjectAltName = @alt_names[ alt_names ]
# 节点主机名与ip
DNS.1 = kafka-1
DNS.2 = kafka-2
DNS.3 = kafka-3
IP.1 = 192.168.100.131
IP.2 = 192.168.100.132
IP.3 = 192.168.100.133
EOF# 签署证书
openssl x509 -req -in kafka.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out kafka.crt \
-days 3650 -extfile san.cnf -extensions req_ext# 检查验证证书
openssl x509 -in kafka.crt -text -noout | grep -A 1 "Subject Alternative Name"ls -ltotal 28
-rw-r--r-- 1 root root 1805 Jan 15 15:54 ca.crt
-rw------- 1 root root 3272 Jan 15 15:54 ca.key
-rw-r--r-- 1 root root   41 Jan 15 15:54 ca.srl
-rw-r--r-- 1 root root 1777 Jan 15 15:54 kafka.crt
-rw-r--r-- 1 root root 1590 Jan 15 15:49 kafka.csr
-rw------- 1 root root 3247 Jan 15 15:49 kafka.key
-rw-r--r-- 1 root root  259 Jan 15 15:51 san.cnf

2. 创建 Keystore

# 将证书和私钥转换为PKCS12文件openssl pkcs12 -export -in kafka.crt -inkey kafka.key -out kafka.p12 -name kafka-cert -CAfile ca.crt -caname root -passout pass:123.com# 使用 keytool 将 kafka.p12 文件导入到 Keystorekeytool -importkeystore \-deststorepass 123.com \-destkeypass 123.com\-destkeystore kafka.keystore.jks \-srckeystore kafka.p12 \-srcstoretype PKCS12 \-srcstorepass 123.com \-alias kafka-cert

3. 创建 Truststore

# 使用keytool创建Truststore并导入CA证书keytool -import \-file ca.crt \-keystore kafka.truststore.jks \-storepass 123.com \-alias root

4. 分发文件

将kafka.truststore.jks 和kafka.keystore.jks 文件分发到其他 kafka 节点

scp kafka.truststore.jks 192.168.100.132:/opt/kafka/pki/scp kafka.keystore.jks  192.168.100.132:/opt/kafka/pki/scp kafka.truststore.jks 192.168.100.133:/opt/kafka/pki/scp kafka.keystore.jks  192.168.100.133:/opt/kafka/pki/

 5. Kafka服务端配置 TLS

# 在Kafka KRaft模式下的server.properties文件中，添加以下配置vim /opt/kafka/config/kraft/server.properties# 修改SSL配置
listeners=SSL://:9092,CONTROLLER://:9093
inter.broker.listener.name=SSL
advertised.listeners=SSL://192.168.100.131:9092,CONTROLLER://192.168.100.131:9093# 新增Keystore配置
ssl.keystore.location=/opt/kafka/pki/kafka.keystore.jks
ssl.keystore.password=123.com
ssl.key.password=123.com
# 新增Truststore配置
ssl.truststore.location=/opt/kafka/pki/kafka.truststore.jks
ssl.truststore.password=123.com
# 客户端连接时启用ssl
ssl.client.auth=required# 重启
systemctl restart kafka

6. 客户端配置 TLS

# 创建客户端配置文件，指定证书信息admin.properties文件内容如下cat > /opt/kafka/config/admin.properties << EOF
security.protocol=SSL
ssl.keystore.location=/opt/kafka/pki/kafka.keystore.jks
ssl.keystore.password=123.com
ssl.truststore.location=/opt/kafka/pki/kafka.truststore.jks
ssl.truststore.password=123.com
ssl.endpoint.identification.algorithm=
ssl.key.password=123.com
EOF# 连接测试## 查看节点信息bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.100.131:9092 --command-config /opt/kafka/config/admin.properties ## 查看topic信息bin/kafka-topics.sh --describe --bootstrap-server 192.168.100.131:9092 --command-config /opt/kafka/config/admin.properties# 生产者生产消息bin/kafka-console-producer.sh --bootstrap-server 192.168.100.131:9092 --topic test --producer.config /opt/kafka/config/admin.properties
>hello boy# 消费者消费消息bin/kafka-console-consumer.sh --bootstrap-server 192.168.10.31:9092 --topic test --from-beginning --consumer.config /opt/kafka/config/admin.propertieshello boy

7. kafka-ui 配置 TLS

# 修改kafka-ui配置文件cat > config.yml << EOF
kafka:clusters:-name: kafka-clusterbootstrapServers: 192.168.100.131:9092,192.168.100.132:9092,192.168.100.133:9092metrics:port: 9997type: JMXproperties:security:protocol: SSLssl:keystore:location: /opt/kafka/pki/kafka.keystore.jkspassword: 123.comssl_endpoint_identification_algorithm: ''ssl:truststorelocation: /opt/kafka/pki/kafka.truststore.jkstruststorepassword: 123.com
EOF# 重启systemctl restart kafka-ui