赞
踩
Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发,支持多语言(如Java、Python、Go等)客户端,它可以水平扩展和具有高吞吐量特性而被广泛使用,并与多类开源分布式处理系统进行集成使用。
Kafka作为一款开源的、轻量级的、分布式、可分区和具备复制备份的、基于ZooKeeper协调管理的分布式流平台的功能强大的消息系统。与传统消息系统相比,Kafka能够更好的处理活跃的流数据,让数据在各个子系统中高性能、低延迟地不停流转。
自0.9.0.0版本开始Kafka社区添加了许多功能用于提高Kafka集群的安全性,Kafka提供SSL或者SASL两种安全策略。SSL方式主要是通过CA令牌实现,此方案主要介绍SASL方式。
验证方式 | kafka版本 | 特点 |
---|---|---|
SASL/PLAIN | 0.10.0.0 | 不能动态添加用户 |
SASL/SCRAM | 0.10.2.0 | 可以动态添加用户 |
SASL/Kerberos | 0.9.0.0 | 需要独立部署验证服务 |
SASL/oauthbearer | 2.0.0 | 需要自己实现接口, 实现token的创建和验证, 需要额外的oauth服务 |
使用SSL加密在代理和客户端之间,代理之间或代理和工具之间传输的数据
SCRAM认证配置的优点:
如果使用PLAIN认证有个问题,就是不能动态新增用户,每次添加用户后,需要重启正在运行的Kafka集群才能生效。
因此,在生产环境中,这种认证方式不符合实际业务场景,不利于后期扩展。然而使用SCRAM认证,可以动态新增用户,添加用户后,可以不用重启正在运行的Kafka集群即可进行鉴权。所以生产环境推荐使用SCRAM+PLAIN搭配的认证方案。
zookeeper官方地址:Apache ZooKeeper
- cat zoo.cfg
-
- tickTime=2000
- initLimit=1
- syncLimit=5
- dataDir=/data/zookeeper/data
- dataLogDir=/data/zookeeper/datalog
- clientPort=2181
- admin.serverPort=8888
- maxClientCnxns=3000
- autopurge.snapRetainCount=3
- autopurge.purgeInterval=24
- server.1=192.168.100.110:2888:3888
- server.2=192.168.100.111:2888:3888
- server.3=192.168.100.112:2888:3888
- 4lw.commands.whitelist=conf,stat,srvr,mntr.envi
- #zk SASL
- authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
- jaasLoginRenew=3600000
- requireClientAuthScheme=sasl
- zookeeper.sasl.client=true
- cat zk_jaas.conf
-
- Server {
- org.apache.zookeeper.server.auth.DigestLoginModule required
- username="admin"
- password="admin123"
- user_kafka="kafka123";
- };
-
- 注意:
-
- admin用户 是zk 集群之间使用的。
- kafka用户 是 broker 与 zk 之间使用的。
- 将上一步添加的 jaas 配置文件添加到zookeeper的环境变量中,zkEnv.sh文件最后添加一行:
-
- vim zkEnv.sh
-
- ZOOBINDIR="${ZOOBINDIR:-/usr/bin}"
- ZOOKEEPER_PREFIX="${ZOOBINDIR}/.."
-
- # 添加如下 新增变量SERVER_JVMFLAGS:
-
- export SERVER_JVMFLAGS="-Djava.security.auth.login.config=../conf/zk_jaas.conf"
- vim /etc/profile
-
- # zk
- export PATH=/home/zookeeper/bin:$PATH
-
- source /etc/profile
-
- # 启动服务
-
- zkServer.sh start && zkServer.sh status
- # 拉取代码
-
- git clone https://github.com/DeemOpen/zkui.git
-
- # 安装maven并配置阿里云源
-
- 从这里下载:https://maven.apache.org/
-
- wget https://dlcdn.apache.org/maven/maven-3/3.9.4/binaries/apache-maven-3.9.4-bin.tar.gz
-
- tar -zxvf apache-maven-3.9.4-bin.tar.gz
-
- echo 'export PATH=/home/apache-maven-3.9.4-bin:$PATH' >>/etc/profile
-
- source /etc/profile
-
- mvn -v
-
- vim settings.xml
-
- 找到</mirrors>节点添加即可
-
- <mirror>
- <id>alimaven</id>
- <name>aliyun maven</name>
- <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
- <mirrorOf>central</mirrorOf>
- </mirror>
- -----------------------------------
-
- # 构建jar包
-
- cd zkui/ && mvn clean install -DskipTests
-
- mkdir /data/zkui
- cp zkui/target/zkui-2.0-SNAPSHOT-jar-with-dependencies.jar /data/zkui/
- cp zkui/config.cfg /data/zkui/
- cd /data/zkui
- nohup java -jar zkui-2.0-SNAPSHOT-jar-with-dependencies.jar &
- tail -n 100 -f nohup.out
- -----------------------------------
-
- http://192.168.100.113:9090/login
- 账号:admin
- 密码:manager
-
- 基本操作
-
- 通过UI界面操作尽量规避掉人为操作的多种不确定性因素导致生产故障
-
- 浏览器访问,下面的用户名和密码提示是config.cfg中loginMessage字段写死的生产中修改掉即可
SASL/SCRAM认证是把凭证(credential)存储在Zookeeper,使用kafka-configs.sh在Zookeeper中创建凭据。对于每个SCRAM机制,必须添加具有机制名称的配置来创建凭证,所以在启动Kafka broker之前需要创建代理间通信的凭据。
这里配置的 Kafka和生产者/消费者之间 采用SASL/PLAIN和SASL/SCRAM两种方式共同完成认证,授权使用ACL方式。PLAIN方式的用户是在jaas文件中写死的,不能动态的添加;SCRAM支持动态的添加用户。
配置SASL/SCRAM认证的第一步,是配置可以连接到kafka集群的用户。本案例创建了3个用户:admin,producer,consumer。kafka_server_admin用户用于broker之间的认证通信,producer用户用于生产者连接kafka,consumer用户用于消费者连接kafka 。
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name admin
-
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name producer
-
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name consumer
-
- Completed Updating config for entity: user-principal 'admin'. #正常情况打印信息这串信息。
-
-
- PS:用户 admin 这里配置admin用户用于实现broker间的通讯。
kafka-configs 脚本是用来设置主题级别参数的。其实,它的功能还有很多。比如在这个例子中,我们使用它来创建 SASL/SCRAM 认证中的用户信息。可以使用下列命令来查看刚才创建的用户数据。
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --describe --entity-type users
-
- #(可以单独指定某个用户 --entity-name producer,如下)
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --describe --entity-type users --entity-name producer
输出示例如下:
Configs for user-principal 'producer' are SCRAM-SHA-512=salt=czVldW1mNDRlcTgzdnJydWxrOXB0YThxbw==,stored_key=Hjrex55CEcLSA2b1n8bvL5CfjziLPHD8EoduoCkNrT0xcVDthoQSi4hvt7szU55pJP/LTbpQkZdV66nueVzKmg==,server_key=OcXwU1KyxBU1tUdlu9ikpfwl+y+ne121iD4amN7zKl8I84KeBZlrwz1IKB1ICFFiAP9XoRRj0QMgHbCfhsL/wA==,iterations=4096,SCRAM-SHA-256=salt=NDJzZmZ0MmhxMDI0N2ZrbXh6bTQwYnZzcA==,stored_key=EIzpjfD4JzItzjeh16UvpQEGyYoZbesR0GRsQB58Als=,server_key=5LAvxe/KjCas7w4GM+KlkT/il99peOozXY/HPZeSF14=,iterations=8192
这段命令包含了 writer 用户加密算法 SCRAM-SHA-256 以及 SCRAM-SHA-512 对应的盐值 (Salt)、ServerKey 和 StoreKey,这些都是 SCRAM 机制的术语。
- ZK客户端命令行查看:
-
- zkCli.sh -server 192.168.100.110:2181
-
- ls /config/users
- 删除用户[producer]的SCRAM证书:
-
- kafka-configs.sh --zookeeper 192.168.100.110:2181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name producer
配置了用户之后,我们需要为 Broker 创建一个对应的 JAAS 文件。在实际场景中,需要为每台单独的物理 Broker 机器都创建一份 JAAS 文件。
Kafka 的 jaas认证配置文件,配置的是登录类,超管密码和管理的帐号密码列表
- vim kafka_server_jaas.conf
-
- KafkaServer {
- org.apache.kafka.common.security.scram.ScramLoginModule required
- username ="admin"
- password="admin123"
- user_admin="admin123"
- user_producer="producer123"
- user_consumer="consumer123";
- };
- KafkaClient {
- org.apache.kafka.common.security.scram.ScramLoginModule required
- username="admin"
- password="admin123"
- user_producer="producer123"
- user_consumer="consumer123";
- };
- Client {
- org.apache.kafka.common.security.scram.ScramLoginModule required
- username="kafka"
- password="kafka123";
- };
-
- -----------------------------------
-
- KafkaServer中usename配置的是kafka服务端使用的账号和密码,后面的user_xxx事预设的普通帐号认证信息。
-
- 中间部分配置的是PLAIN认证方式的账户和密码,其中producer1是账户名,producer123是密码。
- Client配置了broker到Zookeeper的连接用户名密码,这里要和前面zookeeper配置中的zk_jaas.conf.conf 中 user_kafka 的账号和密码相同。
-
- 关于这个文件内容,需要注意以下两点:
-
- 1)不要忘记最后一行和倒数第二行结尾处的分号;
-
- 2)JAAS 文件中不需要任何空格键。
- Kafka 服务配置文件 server.propertis,配置认证协议及认证实现类
-
- # -----------------节点1
-
- cat server.properties
-
- broker.id=0
- listeners=SASL_PLAINTEXT://:9092
- advertised.listeners=SASL_PLAINTEXT://192.168.100.110:9092
- sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
- sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
- security.inter.broker.protocol=SASL_PLAINTEXT
- allow.everyone.if.no.acl.found=false
- authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
- super.users=User:admin
- num.network.threads=3
- num.io.threads=8
- socket.send.buffer.bytes=102400
- socket.receive.buffer.bytes=102400
- socket.request.max.bytes=104857600
- log.dirs=/data/kafka/datalogs
- num.partitions=3
- num.recovery.threads.per.data.dir=1
- offsets.topic.replication.factor=2
- transaction.state.log.replication.factor=1
- transaction.state.log.min.isr=1
- log.flush.interval.messages=10000
- log.flush.interval.ms=1000
- log.retention.hours=168
- log.retention.bytes=1073741824
- log.segment.bytes=1073741824
- log.retention.check.interval.ms=300000
- delete.topic.enable=true
- auto.create.topics.enable=false
- zookeeper.connect=192.168.100.110:2181,192.168.100.111:2181,192.168.100.112:2181
- zookeeper.connection.timeout.ms=60000
- group.initial.rebalance.delay.ms=0
-
- # -------------------节点2
-
- cat server.properties
-
- broker.id=1
- listeners=SASL_PLAINTEXT://:9092
- advertised.listeners=SASL_PLAINTEXT://192.168.100.111:9092
- sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
- sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
- security.inter.broker.protocol=SASL_PLAINTEXT
- allow.everyone.if.no.acl.found=false
- authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
- super.users=User:admin
- num.network.threads=3
- num.io.threads=8
- socket.send.buffer.bytes=102400
- socket.receive.buffer.bytes=102400
- socket.request.max.bytes=104857600
- log.dirs=/data/kafka/datalogs
- num.partitions=3
- num.recovery.threads.per.data.dir=1
- offsets.topic.replication.factor=2
- transaction.state.log.replication.factor=1
- transaction.state.log.min.isr=1
- log.flush.interval.messages=10000
- log.flush.interval.ms=1000
- log.retention.hours=168
- log.retention.bytes=1073741824
- log.segment.bytes=1073741824
- log.retention.check.interval.ms=300000
- delete.topic.enable=true
- auto.create.topics.enable=false
- zookeeper.connect=192.168.100.110:2181,192.168.100.111:2181,192.168.100.112:2181
- zookeeper.connection.timeout.ms=60000
- group.initial.rebalance.delay.ms=0
-
- # -----------------------节点3
- cat server.properties
-
- broker.id=2
- listeners=SASL_PLAINTEXT://:9092
- advertised.listeners=SASL_PLAINTEXT://192.168.100.112:9092
- sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
- sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
- security.inter.broker.protocol=SASL_PLAINTEXT
- allow.everyone.if.no.acl.found=false
- authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
- super.users=User:admin
- num.network.threads=3
- num.io.threads=8
- socket.send.buffer.bytes=102400
- socket.receive.buffer.bytes=102400
- socket.request.max.bytes=104857600
- log.dirs=/data/kafka/datalogs
- num.partitions=3
- num.recovery.threads.per.data.dir=1
- offsets.topic.replication.factor=3
- transaction.state.log.replication.factor=1
- transaction.state.log.min.isr=1
- log.flush.interval.messages=10000
- log.flush.interval.ms=1000
- log.retention.hours=168
- log.retention.bytes=1073741824
- log.segment.bytes=1073741824
- log.retention.check.interval.ms=300000
- delete.topic.enable=true
- auto.create.topics.enable=false
- zookeeper.connect=192.168.100.110:2181,192.168.100.111:2181,192.168.100.112:2181
- zookeeper.connection.timeout.ms=60000
- group.initial.rebalance.delay.ms=0
Kafka 安全认证可以直接通过环境变量 -Djava.security.auth.login.config 设置,修改 Kafka 启动脚本 kafka-start-server.sh 文件最后一行,增加一个参数指向 jaas 配置文件的绝对路径
- vim kafka-server-start.sh
-
- #exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
- exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/path-to-kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"
1) 配置consumer.properties和producer.properties,都要加入以下配置
- ecurity.protocol=SASL_PLAINTEXT
- sasl.mechanism=SCRAM-SHA-512
2) 生产者配置
使用kafka-console-producer.sh脚本测试生产者,由于开启安全认证和授权,此时使用console-producer脚本来尝试发送消息,那么消息会发送失败,原因是没有指定合法的认证用户,因此客户端需要做相应的配置,需要创建一个名为producer.conf的配置文件给producer程序使用。
config目录下创建一个producer.conf的文件,文件内容如下:
cat producer.conf security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.Scra mLoginModule required username="producer" password="producer123";注意:Topic设置写权限
3) 消费者配置
使用kafka-console-consumer.sh脚本测试生产者,由于开启安全认证和授权,因此客户端需要做相应的配置。需要为 consumer 用户创建consumer.conf给消费者程序,同时设置对topic的读权限。
config目录下创建一个consumer.conf的文件,文件内容如下:
vim consumer.conf security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.Scra mLoginModule required username="consumer" password="consumer123";注意:Topic设置读权限。
4) 在生产者和消费者启动脚本中引入JAAS文件
- vim bin/kafka-console-producer.sh
-
- if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
- export KAFKA_HEAP_OPTS="-Xmx512M"
- fi
-
- # 添加这行
- export KAFKA_OPTS="-Djava.security.auth.login.config=/path-to-kafka/config/kafka_server_jaas.conf"
-
- exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
-
-
- # vim bin/kafka-console-consumer.sh
-
- if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
- export KAFKA_HEAP_OPTS="-Xmx512M"
- fi
-
- # 添加这行
- export KAFKA_OPTS="-Djava.security.auth.login.config=/path-to-kafka/config/kafka_server_jaas.conf"
-
- exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
nohup kafka-server-start.sh /path-to-kafka/config/server.properties &
- # 检查启动日志
-
- tail -n 100 -f nohup.out
- zkCli.sh -server localhost:2181
-
- ls /brokers/ids
1) 创建一个测试主题test_topic
kafka-topics.sh --zookeeper 192.168.100.112:2181 --create --replication-factor 3 --partitions 3 --topic test_topic
2)查看创建的Topic
- kafka-topics.sh --list --zookeeper 192.168.100.112:2181 test_topic
-
- kafka-topics.sh --describe --zookeeper 192.168.100.112:2181 --topic test_topic
注:kafka开启认证后,当生产者往Topic写数据需要为主题配置权限(write),即对生产者赋予写的权限。
这里使用producer用户认证授权,通过ACL为producer 用户分配操作test_topic权限(下面两个命令二选一即可):
分配写的权限 kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.100.112:2181 --add --allow- principal User:producer --operation Write --topic 'test_topic'
启动生产者发送消息
特别注意:在生产者生产消息之前如果不设置生产者用户的ACL权限会报错:
kafka-console-producer.sh --broker-list 192.168.100.112:9092 --topic test_topic --producer.config /path-to-kafka/config/producer.conf >hello >world >end
注:kafka开启认证后,当消费者拉取主题的消息时需要为Topic配置权限(read),即对生产者赋予写的权限。
这里使用consumer用户认证授权,通过ACL为consumer用户分配操作test_topic权限,同时需要对消费者组授权(给groupId配权),执行一下2个步骤:
1)分配读的权限 kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.100.112:2181 --add --allow-principal User:consumer--operation Read --topic 'test_topic' 2)分配consumer的权限 kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.100.112:2181 --add --allow- principal User:consumer--consumer --topic test_topic --group test_group启动消费者消费消息:
读取消息前,如果对开启消费者用户读取topic权限,会报错:
#kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.112:9092,192.xxx.xxx.110:9092,192.xxx.xxx.111:9092 --topic test --group test-group --from-beginning --consumer.config /path-to-kafka/config/consumer.conf kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.112:9092 --topic test_topic --group test-group --from-beginning --consumer.config /path-to-kafka/config/consumer.conf hello world end
- 1)创建Topic
- kafka-topics.sh --zookeeper 192.168.100.110:2181 --create --partitions 2 --replication-factor 1 --topic test1
-
-
- 2)生产者端配置
- #添加写权限
- kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.100.110:2181 --add --allow-principal User:producer --operation Write --topic test1
-
-
- #写入消息
- kafka-console-producer.sh --broker-list 192.168.100.112:9092,192.168.100.111:9092,192.168.100.110:9092 --topic test1
- --producer.config /path-to-kafka/config/producer.conf
- >hello
- >world
-
- 3)消费者端配置
- #添加读权限
- kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-
- properties zookeeper.connect=192.168.100.110:2181 --add --allow-principal User:consumer --operation Read --topic test1
-
- #添加消费者组权限
- kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-
- properties zookeeper.connect=192.168.100.110:2181 --add --allow-principal User:consumer --operation Read --group test-group
-
- #拉取消息
- kafka-console-consumer.sh --bootstrap-server 192.168.100.112:9092,192.168.100.110:9092,192.168.100.111:9092
- --topic test1 --from-beginning --consumer.config
- /path-to-kafka/config/consumer.conf --group test-group
- hello
- World
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。