Kafka集群搭建及SASL/SCRAM,ACL權限控制

• • 環境
  • 部署zookeeper集群及安全認證
  • 部署kafka集群及安全認證

環境

系統版本：Centos7.4
kafka版本：kafka_2.12-2.6.0
zookeeper版本：zookeeper-3.6.2
JDK版本：jdk1.8

服務器：
192.168.18.11
192.168.18.12
192.168.18.13

部署目錄
/webapps/kafka
/webapps/zookeeper

部署zookeeper集群及安全認證

zookeeper集群,就是配置文件中相互加入
1.修改配置文件

3臺相同
vi /webapps/zookeeper/conf/zoo.cfg

tickTime=2000 initLimit=10 syncLimit=5 dataDir=/webapps/zookeeper_data/ clientPort=12181 server.1=192.168.18.11:12888:13888 server.2=192.168.18.12:12888:13888 server.3=192.168.18.13:12888:13888 ##SASL安全認證---- authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl jaasLoginRenew=3600000 ##------

2.每臺服務器增加唯一myid
上一步配置文件中目錄 dataDir=/webapps/zookeeper_data/

#192.168.18.11配置 cd /webapps/zookeeper_data echo 1 > myid #192.168.18.12配置 cd /webapps/zookeeper_data echo 2 > myid #192.168.18.13配置 cd /webapps/zookeeper_data echo 3 > myid

3.增加安全認證信息文件zk_server_jaas.conf

3臺服務器都相同,conf文件可放在任意位置
vi /webapps/zookeeper/conf/zk_server_jaas.conf

Server { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin" user_admin="admin"user_kafka="kafka"; };

4.將kafka相關jar包導入zookeeper
將kafka/lib下相關jar包復制到zookeeper/lib下

kafka-clients-1.1.1.jar lz4-java-1.4.1.jar snappy-java-1.1.7.1.jar

5.修改bin/zkEnv.sh,讓zk啟動時讀取到zk_server_jaas.conf

vi /webapps/zookeeper/bin/zkEnv.sh#增加 export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/webapps/zookeeper/conf/zk_server_jaas.conf"

6.啟動
bin/zkServer.sh start

其他命令
停止:bin/zkServer.sh stop
重啟:bin/zkServer.sh restart
狀態:bin/zkServer.sh status

部署kafka集群及安全認證

1.增加安全認證信息文件kafka_server_jaas.conf
3臺服務器相同
KafkaServer 塊中,是kafka設置的認證信息
Client 塊中,是zookeeper的認證信息,用于kafka連接zookeeper

vi /webapps/kafka/config/kafka_server_jaas.conf

KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="admin"; };Client { org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="kafka"password="kafka"; };

2.創建認證
設置config啟動變量
vi /webapps/kafka/bin/kafka-configs.sh

增加

export KAFKA_HEAP_OPTS="-Djava.security.auth.login.config=/webapps/kafka/config/kafka_server_jaas.conf" #創建admin認證,password對應kafka_server_jaas.conf bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin#查看證書 bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --describe --entity-type users --entity-name admin#刪除證書(不操作) bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name admin

3.修改配置文件
vi /webapps/kafka/config/server.properties

#192.168.18.11配置 #設置集群內唯一標識 broker.id=1listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全認證監控服務 advertised.listeners=SASL_PLAINTEXT://192.168.18.11:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.enabled.mechanisms=SCRAM-SHA-512 sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer allow.everyone.if.no.acl.found=true #設置超級用戶 super.users=User:admin#關閉自動創建topic,默認true auto.create.topics.enable=false#log目錄 log.dirs=/webapps/kafka/logs#分區數 num.partitions=3#topic的offset的備份份數 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 #默認備份份數，僅指自動創建的topics default.replication.factor=1#zookeeper連接串 zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181 #192.168.18.12配置#設置集群內唯一標識 broker.id=2listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全認證監控服務 advertised.listeners=SASL_PLAINTEXT://192.168.18.12:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.enabled.mechanisms=SCRAM-SHA-512 sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer allow.everyone.if.no.acl.found=true #設置超級用戶 super.users=User:admin#關閉自動創建topic,默認true auto.create.topics.enable=false#log目錄 log.dirs=/webapps/kafka/logs#分區數 num.partitions=3#topic的offset的備份份數 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 #默認備份份數，僅指自動創建的topics default.replication.factor=1#zookeeper連接串 zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181 #192.168.18.13配置#設置集群內唯一標識 broker.id=3listeners=SASL_PLAINTEXT://0.0.0.0:9092#安全認證監控服務 advertised.listeners=SASL_PLAINTEXT://192.168.18.13:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.enabled.mechanisms=SCRAM-SHA-512 sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer allow.everyone.if.no.acl.found=true #設置超級用戶 super.users=User:admin#關閉自動創建topic,默認true auto.create.topics.enable=false#log目錄 log.dirs=/webapps/kafka/logs#分區數 num.partitions=3#topic的offset的備份份數 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 #默認備份份數，僅指自動創建的topics default.replication.factor=1#zookeeper連接串 zookeeper.connect=192.168.18.11:12181,192.168.18.12:12181,192.168.18.13:12181

4.配置kafka啟動環境變量
vi /webapps/kafka/bin/kafka-server-start.sh

#增加 export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G -Djava.security.auth.login.config=/webapps/kafka/config/kafka_server_jaas.conf"

5.啟動
bin/kafka-server-start.sh -daemon config/server.properties

6.增加admin認證文件
vi /webapps/kafka/config/admin.conf

security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";

7.創建topic

#創建topic bin/kafka-topics.sh --create --bootstrap-server 192.168.18.11:9092 --replication-factor 1 --partitions 1 --topic mytest --command-config config/admin.conf#查看所有topic bin/kafka-topics.sh --list --bootstrap-server 192.168.18.11:9092 --command-config config/admin.conf

8.操作用戶

#新增用戶,新建用戶mytest bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name mytest#更新用戶,更新mytest的密碼為mytest bin/kafka-configs.sh --zookeeper 192.168.18.11:12181 --alter --add-config 'SCRAM-SHA-512=[password=mytest]' --entity-type users --entity-name mytest

9.設置權限

#讀取權限,設置用戶mytest的消費者權限 bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --add --allow-principal User:"mytest" --consumer --topic 'mytest' --group '*'#寫入權限,設置用戶mytest的生產者權限 bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --add --allow-principal User:"mytest" --producer --topic 'mytest'#查看權限 bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.18.11:12181 --list

10.測試

新建mytest認證文件

vi /webapps/kafka/config/mytest.conf

security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="mytest" password="mytest";

2臺機器分別執行生產者和消費者-------------生產者生產消息,看消費者是否能接收到

#生產者 bin/kafka-console-producer.sh --broker-list 192.168.18.11:9092 --topic mytest --producer.config config/mytest.conf#消費者 bin/kafka-console-consumer.sh --bootstrap-server 192.168.18.11:9092 --topic mytest --consumer.config config/mytest.conf

11.查看groupid和offset

#查看所有groupid bin/kafka-consumer-groups.sh --bootstrap-server 192.168.18.11:9092 --list --command-config config/admin.conf#查看offset bin/kafka-consumer-groups.sh --bootstrap-server 192.168.18.11:9092 --describe --group test-consumer-group --command-config config/admin.conf

12.查看消息內容

#查看消息內容: 消息追蹤 bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /webapps/kafka/logs/mytest-0/00000000000000000000.log --print-data-log

總結

以上是生活随笔為你收集整理的Kafka集群搭建及SASL/SCRAM,ACL权限控制的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。