kafka 安全認證,kafka 支持的認證機制包括SSL(TLS)或SASL穴豫,本文基于SASL進行配置铺峭。
1、kafka 服務(wù)端配置
- 1 在config目錄下新建文件 kafka_server_jaas.conf,內(nèi)容如下:
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_producer="producer"
user_consumer="consumer";
};
注:結(jié)尾必須加封號(;),要不然不生效
- 在config目錄下新建文件 kafka_zookeeper_jaas.conf,內(nèi)容如下:
Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin";
};
- 在config目錄下新建文件 kafka_client_jaas.conf,內(nèi)容如下:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="test"
password="test";
};
注:該配置是通過kafka-console-producer.bat弹囚、kafka-console-consumer.bat 手動輸入命令時會使用到,同時也需要在.bat文件中進行配置领曼,加載該文件
- 在server.properties 中添加以下內(nèi)容:
#安全認證
listeners=SASL_PLAINTEXT://127.0.0.1:9092
#使用的認證協(xié)議
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL機制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份驗證的類
#kafka 3.0版本后棄用SimpleAclAuthorizer驗證鸥鹉,改為kafka.security.authorizer.AclAuthorizer
#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
#如果沒有找到ACL(訪問控制列表)配置,則允許任何操作庶骄。
#allow.everyone.if.no.acl.found=true
#需要開啟設(shè)置超級管理員, 開啟以下命令
super.users=User:admin
- 在zookeeper.properties 添加如下內(nèi)容:
# 安全身份認證
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
- 在kafka-server-start.bat 中添加如下內(nèi)容:
IF ["%KAFKA_OPTS%"] EQU [""] (
set KAFKA_OPTS=-Djava.security.auth.login.config=file:%~dp0../../config/kafka_server_jaas.conf
)
- 在zookeeper-server-start.bat 中添加如下內(nèi)容:
set KAFKA_OPTS=-Djava.security.auth.login.config=file:%~dp0../../config/kafka_zookeeper_jaas.conf
- 依次啟動zookee服務(wù)毁渗,kafka服務(wù)。命令如下:
zkServer.cmd
kafka-server-start.bat ..\..\config\server.properties
2单刁、springboot配置(生產(chǎn)者/消費者):
kafka連接地址
spring.kafka.bootstrap-servers=localhost:9092
#kafka生產(chǎn)者配置
#序列化key的類
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
#序列化value的類
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.StringSerializer
#kafk消費者配置
#是否開啟自動提交 手動提交也需要設(shè)置ack-mode
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=MANUAL
#kafka 安全認證
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
spring.kafka.properties.security.protocol=SASL_PLAINTEXT
#kafka生產(chǎn)者配置
#序列化key的類
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
#序列化value的類
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.StringSerializer
#kafka 安全認證
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=”admin“ password=“admin”;
spring.kafka.properties.security.protocol=SASL_PLAINTEXT
注:最后的封號不能少
propertiest文件安全認證配置:
##kafka 安全認證
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="123456";
spring.kafka.properties.security.protocol=SASL_PLAINTEXT
yml文件安全認證配置
spring:
kafka:
bootstrap-servers:
- localhost:9092
properties:
sasl:
mechanism: PLAIN
jaas:
config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="123456";'
security:
protocol: SASL_PLAINTEXT
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
enable-auto-commit: true
group-id: T1
producer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
參照官網(wǎng):https://kafka.apache.org/documentation/#brokerconfigs