【Kafka SASL认证】_kafkaproducer sasl

条件：

• 注：先安装Zookeeper和Kafka
• 链接：https://blog.csdn.net/ZhongYuxuanG

Kafka配置SASL认证：

1.创建文件中配置用户

①.kafka/config/下创建kafka_client_jaas.conf文件：

KafkaClient {  
org.apache.kafka.common.security.plain.PlainLoginModule required  
  username="admin"  
  password="admin";  
};
1
2
3
4
5

②.kafka/config/下创建kafka_server_jaas.conf文件：

KafkaServer {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="admin"
        user_admin="admin"
        user_test1="123456"
        user_test2="1234567";
};

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

Client {
	org.apache.kafka.common.security.plain.PlainLoginModule required
		username="admin"
		password="admin";
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

③.kafka/config/下创建kafka_zoo_jaas.conf文件：

ZKServer{
	org.apache.kafka.common.security.plain.PlainLoginModule required
		username="admin"
		password="admin"
		user_admin="admin";
};
1
2
3
4
5
6

④.kafka/config/consumer.properties文件，添加下面内容：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
1
2

⑤.kafka/config/producer.properties文件，添加下面内容：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
1
2

⑥.kafka/config/zookeeper.properties文件，添加下面内容：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
1
2
3

⑦.kafka/config/server.properties文件，添加下面内容：

listeners=SASL_PLAINTEXT://ip:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
delete.topic.enable=true
auto.create.topics.enable=false
1
2
3
4
5
6
7
8

2.修改Kafka的bin文件(注意：里面有自己的路径需要修改)

①.修改zookeeper-server-start.sh文件：

export KAFKA_OPTS="-Djava.security.auth.login.config=/自己的路径/kafka_2.11-0.11.0.0/config/kafka_zoo_jaas.conf -Dzookeeper.sasl.serverconfig=ZKServer"
1

②.修改kafka-server-start.sh文件：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/自己的路径/kafka_2.11-0.11.0.0/config/kafka_server_jaas.conf"
1

③.修改kafka-console-producer.sh文件：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/自己的路径/kafka_2.11-0.11.0.0/config/kafka_client_jaas.conf"
1

④.修改kafka-console-consumer.sh文件：

export KAFKA_OPTS=" -Djava.security.auth.login.config=/自己的路径/kafka_2.11-0.11.0.0/config/kafka_client_jaas.conf"
1

3.启动

①.启动zookeeper：

bin/zookeeper-server-start.sh config/zookeeper.properties
1

②.启动kafka：

bin/kafka-server-start.sh config/server.properties
1

4.生产、消费

①.生产：

bin/kafka-console-producer.sh --broker-list ip:9092 --topic test --producer.config config/producer.properties
1

①.消费：

bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --topic test --from-beginning --consumer.config config/consumer.properties
1