如何配置kafka的SSL链接访问
主要介绍kafka和client端之间的SSL连接配置。
- 配置kafka端, 即server端
配置conf/server.properties内容：
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.client.auth=required
listeners=PLAINTEXT://kafka1:9092,SSL://kafka1:9093
advertised.listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
# security.inter.broker.protocol=SSL
几点说明：
- ssl.truststore/ssl.keystore
这个没啥好说的，就是SSL需要的证书信息。 - ssl.client.auth=required
这个指示是否配置成双向验证(2-way authentication)。在通常情况下(即如果没有设置ssl.client.auth=required)，kafka实行的是单向验证，也就是客户端验证kafka服务器，而kafka服务器不验证客户端。 - listeners/advertised.listeners
为什么会有两个，实际上kafka支持多端口的访问发布，可以两个或者以上，每一个端口实行不同的访问接口协议。 - security.inter.broker.protocol=SSL
指示kafka之间(broker-to-broker)的内部通信使用那种协议，缺省是PLAINTEXT。这个参数要和listeners/advertised.listeners一起来看，如果listeners/advertised.listeners只配置了单个端口协议，那么security.inter.broker.protocol必须根据实际的协议设置。例如：
4.1 如果listeners=PLAINTEXT://kafka1:9092
那么security.inter.broker.protocol就不可以设置成SSL
4.2 如果 listeners=SSL://kafka1:9093
那么security.inter.broker.protocol就必须设置成SSL
4.3 如果listeners=PLAINTEXT://kafka1:9092,SSL://kafka1:9093
那么security.inter.broker.protocol就可以选择设置成PLAINTEXT或者SSL，依赖于配置的brokers-to-brokers访问端口是9092还是9093。
通常情况下，我们建议不把security.inter.broker.protocol=SSL，也就是说让kafka的内部通信使用PLAINTEXT，因为SSL毕竟影响性能的。而让client-to-broker的通道配置成SSL，鉴于安全性的考虑，而且最好配置成双向认证。如果是在docker里面运行，只需要把SSL端口映射到外面给client-to-brokers访问，而内部brokers-to-brokers的PLAINTEXT端口根本不映射出来。
kafka服务器端的配置就这些内容；如果是从docker-compose来配置的话，可以写到docker-compose.yaml文件里：
environment:
- KAFKA_BROKER_ID=1
- KAFKA_LOG_DIRS=/work/klog
- KAFKA_DATA_DIRS=/work/kdata
- KAFKA_ZOOKEEPER_CONNECT=zookeeper1:2181,zookeeper2:2181
- KAFKA_LISTENERS=PLAINTEXT://kafka1:9092,SSL://kafka1:9093
- KAFKA_ADVERTISED.LISTENERS=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
#- KAFKA_SECURITY.INTER.BROKER.PROTOCOL=SSL
- KAFKA_SSL.CLIENT.AUTH=required
- KAFKA_SSL.KEYSTORE.LOCATION=/work/conf/server.keystore.jks
- KAFKA_SSL.KEYSTORE.PASSWORD=test1234
- KAFKA_SSL.KEY.PASSWORD=test1234
- KAFKA_SSL.TRUSTSTORE.LOCATION=/work/conf/truststore.jks
- KAFKA_SSL.TRUSTSTORE.PASSWORD=test1234
- 配置客户端(client)
以kafka-console-producer/kafka-console-consumer为例。
创建一个client-ssl.properties文件，如下：
$ cat client-ssl.properties
bootstrap.servers=kafka1:9093,kafka2:9093
security.protocol=SSL
ssl.truststore.location=/work/conf/truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/work/conf/client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
如下格式执行命令：
- Producer
/opt/kafka/bin/kafka-console-producer.sh --broker-list kafka1:9093,kafka2:9093 \
--topic test-topic \
--producer.config client-ssl.properties
- Consumer
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka1:9093,kafka2:9093 \
--topic test-topic --partition 0 \
--consumer.config client-ssl.properties \
--from-beginning
这样就可以了。
