Kafka配置SSL认证

Posted 2023-01-14 justry_deng

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka配置SSL认证相关的知识，希望对你有一定的参考价值。

Kafka配置SSL认证

使用kafka自带的消费者生产者测试一下

我是一只小小小小鸟~嗷！嗷！

Kafka配置SSL认证

注：其实对于本节内容来说，有SSL的服务端证书就够了。

第一步：修改kafka安装目录下config目录下的server.properties文件

############################# Server Basics #############################
# SSL认证配置
# 如果配置了SSL认证，那么原来的port和advertised.listeners可以注释掉了
listeners=SSL://kafka-single:9095
advertised.listeners=SSL://kafka-single:9095
ssl.keystore.location=/usr/ca/server/server.keystore.jks
ssl.keystore.password=ds1994
ssl.key.password=ds1994
ssl.truststore.location=/usr/ca/trust/server.truststore.jks
ssl.truststore.password=ds1994
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS 
ssl.truststore.type=JKS 
# kafka2.0.x开始，将ssl.endpoint.identification.algorithm设置为了HTTPS，即:需要验证主机名
# 如果不需要验证主机名，那么可以这么设置 ssl.endpoint.identification.algorithm=即可
ssl.endpoint.identification.algorithm=HTTPS
# 设置内部访问也用SSL，默认值为security.inter.broker.protocol=PLAINTEXT
security.inter.broker.protocol=SSL
broker.id=0
############################# Socket Server Settings #############################
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
############################# Log Basics #############################
log.dirs=/usr/data/kafka
num.partitions=1
num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings  #############################
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Retention Policy #############################
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
############################# Group Coordinator Settings #############################
group.initial.rebalance.delay.ms=0

注：SSL配置最好写在配置文件的最上面，否者可能导致Kafka配置SSL失败。

注：如果设置的内部broker的通讯协议PLAINTEXT，那么监听PLAINTEXT的时候就需要作相应的配置
listeners=PLAINTEXT://host.name:port,SSL://host.name:port。

第二步(可选)：如果配置SSL之前，存在Kafka数据，那么建议重新换一个位置来
存放数据；如果确保之前的数据已经没什么用了，也可以直接删除
之前的数据

注：之所以是这个文件夹，是因为是我们在安装kafka的时候指定的数据目录，可
详见https://blog.csdn.net/justry_deng/article/details/88381595。

第三步：重启kafka

# 后台启动zookeeper
/var/local/kafka/bin/zookeeper-server-start.sh /var/local/kafka/config/zookeeper.properties > /usr/data/zookeeper.log 2>&1 &
# 前台启动kafak
/var/local/kafka/bin/kafka-server-start.sh /var/local/kafka/config/server.properties

注：启动kafka时最好使用进程独占一个shell的方式前台启动，这样能非常直观的查看启动kafka是否成功。

第四步：使用Linux自带的openssl测试一下，验证我们配置的ssl有效

openssl s_client -debug -connect kafka-single:9095 -tls1

注：kafka-single为/etc/hosts文件中配置的host名。

弹出以下内容，说明成功了:

……

到此为止，就说明初步成功了！

使用kafka自带的消费者生产者测试一下

准备工作：先创建一个主题

# 创建主题
/var/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topicOne
# 查看所有主题
/var/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181

执行效果如图：

消费者：

首先创建一个SSL下的消费者配置文件c.properties，编辑其内容为：

security.protocol=SSL
group.id=test-group
ssl.truststore.location=/usr/ca/trust/server.truststore.jks
ssl.truststore.password=ds1994
ssl.keystore.password=ds1994
ssl.keystore.location=/usr/ca/server/server.keystore.jks

启动消费者：

/var/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka-single:9095 --topic topicOne --from-beginning --consumer.config /var/local/kafka/config/c.properties

生产者：

首先创建一个SSL下的消费者配置文件p.properties，编辑其内容为：

bootstrap.servers=kafka-single:9095
security.protocol=SSL
ssl.truststore.location=/usr/ca/trust/server.truststore.jks
ssl.truststore.password=ds1994   
ssl.keystore.password=ds1994
ssl.keystore.location=/usr/ca/server/server.keystore.jks

启动生产者：

/var/local/kafka/bin/kafka-console-producer.sh --broker-list kafka-single:9095 --topic topicOne --producer.config /var/local/kafka/config/p.properties