做轻时尚的网站,豪华跑车网站建设,海外域名平台,计算机技术员网站建设怎么作者 乐维社区#xff08;forum.lwops.cn#xff09;许远 在构建现代的分布式系统时#xff0c;确保数据传输的安全性至关重要。Apache Kafka 和 Zookeeper 作为流行的分布式消息队列和协调服务#xff0c;提供了SSL#xff08;Secure Sockets Layer#xff09;认证机制forum.lwops.cn许远 在构建现代的分布式系统时确保数据传输的安全性至关重要。Apache Kafka 和 Zookeeper 作为流行的分布式消息队列和协调服务提供了SSLSecure Sockets Layer认证机制以增强数据传输过程中的安全性。 本文将详细介绍从生成SSL证书到配置服务端和客户端的全过程确保数据在传输过程中得到充分的保护。
一、配置Kafka账号密码 1、首先需要修改kafka配置文件vim /asop/kafka/kafka_2.11-2.1.0/config/server.properties
broker.id0 listenersSASL_PLAINTEXT://:9092 advertised.listenersSASL_PLAINTEXT://10.176.31.137:9092 num.network.threads3 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 log.dirs/asop/kafka/logs num.partitions1 num.recovery.threads.per.data.dir1 offsets.topic.replication.factor1 transaction.state.log.replication.factor1 transaction.state.log.min.isr1 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 zookeeper.connectlocalhost:2181 zookeeper.connection.timeout.ms6000 group.initial.rebalance.delay.ms0
#使用的认证协议 security.inter.broker.protocolSASL_PLAINTEXT #SASL机制 sasl.enabled.mechanismsPLAIN sasl.mechanism.inter.broker.protocolPLAIN #完成身份验证的类 authorizer.class.namekafka.security.auth.SimpleAclAuthorizer #如果没有找到ACL访问控制列表配置则允许任何操作。 allow.everyone.if.no.acl.foundfalse #需要开启设置超级管理员,设置visitor用户为超级管理员 super.usersUser:visitor
2、其次为server创建登录验证文件,可以根据自己爱好命名文件如vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf 文件内容如下
KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username“visitor” password“qaz123” user_visitor“qaz123”; };
3、然后修改kafka安装目录vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh在文件最上面添加变量
export KAFKA_OPTS -Djava.security.auth.login.config/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf 4、接下来为consumer和producer创建登录验证文件可以根据爱好命名文件如kafka_client_jaas.conf文件内容如下如果是程序访问如springboot访问可以不配置 vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf
KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username“visitor” password“qaz123”; };
5、在consumer.properties和producer.properties里分别加上如下配置 vim /asop/kafka/kafka_2.11-2.1.0/config/consumer.properties vim /asop/kafka/kafka_2.11-2.1.0/config/producer.properties
security.protocolSASL_PLAINTEXT sasl.mechanismPLAIN
6、修改kafka安装目录bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh在文件最上面添加变量 vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh
export KAFKA_OPTS -Djava.security.auth.login.config/asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf 7、分别启动zookeeper和kafka,至此服务端kafka用户登录验证配置完成先关闭kafka后关闭zookeeper
关闭服务kafka /asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties
启动服务kafka #/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties
关闭服务zookeeper-3.4.13 /asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg 启动服务zookeeper-3.4.13 /asop/zk/zookeeper-3.4.13/bin/zkServer.sh start /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
8、创建及查看主题
/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh --broker-list 10.176.31.137:9092 --topic cmdb --producer-property security.protocolSASL_PLAINTEXT --producer-property sasl.mechanismPLAIN 接收消息
/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server 10.176.31.137:9092 --topic cmdb --from-beginning --consumer-property security.protocolSASL_PLAINTEXT --consumer-property sasl.mechanismPLAIN
二、zk和kafka配置ssal账号密码
Zookeeper 配置 SASL 1.1 新建 zoo_jaas.conf 文件 zoo_jaas.conf 文件名、文件所在路径没有特殊要求一般放置在${ZOOKEEPER_HOME}/conf目录下vim /asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf
Server { org.apache.kafka.common.security.plain.PlainLoginModule required username“admin” password“admin12” user_kafka“kafka123”; }; Server.username、Server.password为 Zookeeper 内部通信的用户名和密码因此保证每个 zk 节点该属性一致即可Server.user_xxx 中 xxx 为自定义用户名用于 zkClient 连接所使用的用户名和密码即为 kafka 创建的用户名1.2 配置 /asop/zk/zookeeper-3.4.13/conf/zoo.cfg 文件 authProvider.1org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthSchemesasl jaasLoginRenew3600000 zookeeper.sasl.clienttrue
zookeeper.sasl.client 设置为 true开启客户端身份验证否则zoo_jaas.conf中配置的用户名将不起作用客户端仍然可以无 jaas 文件连接只是带有 WARNNING 而已
1.3 导入依赖包 因为使用的权限验证类为org.apache.kafka.common.security.plain.PlainLoginModule所以需要 kafka 相关 jar 包新建文件夹 zk_sasl_lib如下从kafka/lib目录下复制以下几个jar包到zookeeper的lib和新建的zk_sasl_lib目录下
kafka-clients-2.4.1.jar lz4-java-1.6.0.jar slf4j-api-1.7.28.jar slf4j-log4j12-1.7.28.jar snappy-java-1.1.7.3.jar
mkdir /asop/zk/zookeeper-3.4.13/zk_sasl_lib cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/lib/ cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/lib/ cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/ cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/ cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/lib/ cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/ chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/
1.4 修改 zkEnv.sh 文件/asop/zk/zookeeper-3.4.13/bin/zkEnv.sh 修改前如果没有就直接加
export SERVER_JVMFLAGS“-Xmx${ZK_SERVER_HEAP}m $SERVER_JVMFLAGS”
修改后
for jar in /asop/zk/zookeeper-3.4.13/zk_sasl_lib/*.jar; do CLASSPATH“ j a r : jar: jar:CLASSPATH” done
export SERVER_JVMFLAGS -Djava.security.auth.login.config/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf
重启 Zookeeper 服务即可
关闭服务zookeeper-3.4.13 /asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg 启动服务zookeeper-3.4.13 /asop/zk/zookeeper-3.4.13/bin/zkServer.sh start /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
Kakfa 配置 SASL 2.1 新建 kafka_server_jaas.conf 文件 kafka_server_jaas.conf 文件名和存放路径没有要求一般放置在${KAFKA_HOME}/config目录下/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf
KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username“visitor” password“qaz123” user_visitor“qaz123”; }; Client{ org.apache.kafka.common.security.plain.PlainLoginModule required username“kafka” password“kafka123”; }; KafkaServer.username、KafkaServer.password 为 broker 内部通信的用户名密码同上KafkaServer.user_xxx 其中 xxx 必须和 KafkaServer.username 配置的用户名一致密码也一致 KafkaServer.user_producer、KafkaServer.user_consumer 为了之后的 ACL 做准备达到消费者生产者使用不同账号且消费者账号只能消费数据生产者账号只能生产数据 Client.username、Client.password 填写 Zookeeper 中注册的账号密码用于 broker 与 zk 的通信若 zk 没有配置 SASL 可以忽略、若 zookeeper.sasl.client 为 false 也可以忽略只是带有日志如下
[2021-06-29 17:14:30,204] WARN SASL configuration failed: javax.security.auth.login.LoginException: No JAAS configuration section named ‘Client’ was found in specified JAAS configuration file: ‘/Users/wjun/env/kafka/config/kafka_server_jaas.conf’. Will continue connection to Zookeeper server without SASL authentication, if Zookeeper server allows it. (org.apache.zookeeper.ClientCnxn)
2.2 修改 server.properties 文件 broker.id0 listenersSASL_PLAINTEXT://:9092 advertised.listenersSASL_PLAINTEXT://192.168.157.198:9092 num.network.threads3 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 log.dirs/asop/kafka/logs num.partitions1 num.recovery.threads.per.data.dir1 offsets.topic.replication.factor1 transaction.state.log.replication.factor1 transaction.state.log.min.isr1 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 zookeeper.connect127.0.0.1:2181 zookeeper.connection.timeout.ms6000 group.initial.rebalance.delay.ms0
#使用的认证协议 security.inter.broker.protocolSASL_PLAINTEXT #SASL机制 sasl.enabled.mechanismsPLAIN sasl.mechanism.inter.broker.protocolPLAIN #完成身份验证的类 authorizer.class.namekafka.security.auth.SimpleAclAuthorizer #如果没有找到ACL访问控制列表配置则允许任何操作。 allow.everyone.if.no.acl.foundfalse #需要开启设置超级管理员,设置visitor用户为超级管理员 super.usersUser:visitor
其中 localhost 需要修改成 IP地址
super.users 配置超级用户该用户不受之后的 ACL 配置影响
2.3 修改启动脚本 修改 kafka-server-start.sh 文件使之加载到 kafka_server_jaas.conf 文件/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh
修改前
if [ “x$KAFKA_HEAP_OPTS” “x” ]; then export KAFKA_HEAP_OPTS“-Xmx1G -Xms1G” fi
修改后 先在首行加这一行,如果有了就不用加了export KAFKA_OPTS -Djava.security.auth.login.config/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf if [ “x$KAFKA_HEAP_OPTS” “x” ]; then export KAFKA_HEAP_OPTS“-Xmx1G -Xms1G -Djava.security.auth.login.config/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf” fi
设置zookeeper的ACL规则 /asop/zk/zookeeper-3.4.13/bin/zkCli.sh #进入zk的命令行模式
addauth digest admin:admin12 #切换登陆用户超级管理员是在zk的配置文件/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf里面
setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka123:cdrwa #(设置可以登陆的IP和用户账号密码admin是上面的zk的配置文件里面定义的管理员Kafka用户是/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf文件里面的定义的kafka连接zk的用户 Client下面的)
addauth digest kafka:kafka123 #再切换为kafka用户再设置一次acl setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka123:cdrwa
注意如果要加白名单IP或者用户要在原来的基础上加不然会覆盖 setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka123:cdrwa,auth:admin:admin12:cdrwa,ip:1.1.1.1
需要恢复权限不设置acl的话就运行 setAcl / world:anyone:cdrwa
重启 kafka 服务即可
关闭服务kafka #/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties
启动服务kafka #/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties
至此完成kafka与zookeeper配置ssl认证。更多运维技巧欢迎关注乐维社区更多运维问题也欢迎留言提问。
