程序版本

scala-2.12.3.tgz自带zookeeper  安装步骤

1 切换到安装目录下，新建配置文件
 

cd /usr/local/kafka/kafka_2.12-2.5.0

touch kafka_server_jaas.conf

vi kafka_server_jaas.conf

 

输入一下内容

KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-test"
    user_admin="admin-test"
    user_producer="prod-test"
    user_consumer="cons-test";
};

定义了关键字KafkaServer字段用于指定服务端登录配置。该配置通过org.apache.org.apache.kafka.common.security.plain.PlainLoginModule由指定采用PLAIN 机制， 定义了用户， 用户通
过usemame 和password 指定该代理与集群其他代理初始化连接的用户名和密码， 通过“ user_ "为前缀后接用户名方式创建连接代理的用户名和密码，例如， user_producer="prod-test" 是指用户名为producer， 密码为prod-test。

2 修改kafka启动时的配置文件，server.properties
 

vi /usr/local/kafka/kafka_2.12-2.5.0/config/server.properties

修改内容如下，在文末添加如下内容：
注意点：192.168.2.122是我当前主机ip,9092是kafka通信端口，其他的地方保持一致

看下server.properties文件中是否已经含有，不要配置重复项

listeners=SASL_PLAINTEXT://192.168.2.122:9092
advertised.listeners=SASL_PLAINTEXT://192.168.2.122:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true

3 修改启动脚本

vi /usr/local/kafka/kafka_2.12-2.5.0/bin/kafka-server-start.sh

找到  export KAFKA_HEAP_OPTS,添加jvm 参数kafka_server_jaas.conf文件
 

-Djava.security.auth.login.config=/usr/local/kafka/kafka_2.12-2.5.0/kafka_server_jaas.conf

 添加后效果如下：

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/usr/local/software/kafka/kafka_server_jaas.conf"
fi

4 验证

1 创建client认证文件，此文件是后面console的生产者和消费者使用，同样此文件不适宜添加任何多余的内容，包括注释：

cd /usr/local/kafka/kafka_2.12-2.5.0

touch kafka_client_jaas.conf

输入内容

KafkaClient {
	org.apache.kafka.common.security.plain.PlainLoginModule required
	username="producer"
	password="prod-test";
};

2添加kafka-console-producer.sh认证文件路径，后面启动生产者测试时使用：

vi bin/kafka-console-producer.sh

添加参数

-Djava.security.auth.login.config=/usr/local/kafka/kafka_2.12-2.5.0/kafka_client_jaas.conf

 

3 添加kafka-console-consumer.sh认证文件路径，后面启动消费者测试时使用：

vi bin/kafka-console-consumer.sh

同样添加

-Djava.security.auth.login.config=/usr/local/kafka/kafka_2.12-2.5.0/kafka_client_jaas.conf

4 修改 config/producer.properties，在配置最后加入以下两行内容：

vi config/producer.properties

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

5 修改 config/consumer.properties，要添加的内容和producer的内容一样

6 开启生产者

bin/kafka-console-producer.sh --broker-list 192.168.2.122:9092 --topic test --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

7 开启消费者

bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.122:9092 --topic test --from-beginning --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN