【kafka2.4.1】kafka增加身份认证_kafka开启认证

kafka默认部署情况下是没有身份认证。
基础环境准备：
1、版本 kafka 2.4.1 ，zookeeper3.5.7(自行下载)

2、单节点 kafka
配置开始（ 只需要配置kafka 即可，无需配置zk）：
1、进入kafka目录，修改kafka配置文件，修改config/server.properties文件，并添加如下内容

vi config/server.properties
1

#配置zookeeper管理kafka的路径
zookeeper.connect=localhost:2181
#配置kafka的监听端口（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
listeners=SASL_PLAINTEXT://:9092
#把kafka的地址端口注册给zookeeper，如果是远程访问要改成外网IP（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
advertised.listeners=SASL_PLAINTEXT://你外网的ip:9092
#使用的认证协议（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer#如果没有找到ACL（访问控制列表）配置，则允许任何操作。
allow.everyone.if.no.acl.found=false
#需要开启设置超级管理员,设置visitor用户为超级管理员
super.users=User:visitor
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2、在kafka目录下config目录，创建配置文件kafka_server_jaas.conf，填入如下内容。

KafkaServer {
	org.apache.kafka.common.security.plain.PlainLoginModule 			
	requiredusername="visitor"
	password="qaz@123"
	user_visitor="qaz@123";
};
1
2
3
4
5
6

3、在当前kafka目录下，修改kafka启动脚本

vi bin/kafka-server-start.sh
1

在文件最上面添加变量

export KAFKA_OPTS=" -Djava.security.auth.login.config=/xxx/config/kafka_server_jaas.conf"
1

4、启动zk、kafka

zk启动
./zkServer.sh start
kafka 启动
nohup bin/kafka-server-start.sh -daemon config/server.properties  >> logs/server.log  2>&1 &
1
2
3
4

5、springboot 连接kafka

spring:
  kafka:
    # 指定kafka 代理地址，可以多个
    bootstrap-servers: xxx.xxx.xxx.xxx:9092 
    producer: # 生产者
      retries: 1 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      # 每次批量发送消息的数量
      batch-size: 16384
      buffer-memory: 33554432
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      #修改最大向kafka推送消息大小
      properties:
        max.request.size: 52428800
    consumer:
      #手动提交offset保证数据一定被消费
      enable-auto-commit: false
      #指定从最近地方开始消费(earliest)
      auto-offset-reset: latest
      #消费者组
      #group-id: dev
    properties:
      security:
        protocol: SASL_PLAINTEXT
      sasl:
        mechanism: PLAIN
        jaas:
          config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="visitor" password="qaz@123";'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

文章参考：http://wed.xjx100.cn/news/216531.html?action=onClick