Kafka如何配置用户名密码访问

1 软件版本
kafka_2.12-2.4.0.tgz（带zookeeper）http://archive.apache.org/dist/kafka/2.4.0/kafka_2.12-2.4.0.tgz

2 kafka服务端部署
2.1 将安装包上传到服务器，并解压

1. tar zxvf kafka_2.12-2.4.0.tgz -C /data
   
2. 
   
3. mv kafka_2.12-2.4.0 kafka
   

复制代码

2.2 修改kafka配置文件 server.properties
vim /data/kafka/config/server.properties:

1. ############################# Server Basics #############################
   
2. 
   
3. # The id of the broker. This must be set to a unique integer for each broker.
   
4. 
   
5. broker.id=0
   
6. host.name=10.7.2.201
   
7. 
   
8. listeners=SASL_PLAINTEXT://10.7.2.201:9092
   
9. advertised.listeners=SASL_PLAINTEXT://10.7.2.201:9092
   
10. 
    
11. security.inter.broker.protocol=SASL_PLAINTEXT
    
12. sasl.enabled.mechanisms=PLAIN
    
13. sasl.mechanism.inter.broker.protocol=PLAIN
    
14. authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
    
15. allow.everyone.if.no.acl.found=true
    
16. 
    

复制代码

2.3 添加SASL配置文件
在kafka安装目录下创建一个 kafka_server_jaas.conf 文件
vim /data/kafka/kafka_server_jaas.conf:

1. KafkaServer {
   
2.   org.apache.kafka.common.security.plain.PlainLoginModule required
   
3.     username="admin"
   
4.     password="adminpasswd"
   
5.     user_admin="adminpasswd"
   
6.     user_producer="producerpwd"
   
7.     user_consumer="consumerpwd";
   
8. };
   
9. 
   

复制代码

说明：该配置通过org.apache.org.apache.kafka.common.security.plain.PlainLoginModule由指定采用PLAIN机制，定义了用户。
    usemame和password指定该代理与集群其他代理初始化连接的用户名和密码
    "user_"为前缀后接用户名方式创建连接代理的用户名和密码，例如，user_producer=“producerpwd” 是指用户名为producer，密码为producerpwd
    username为admin的用户，和user为admin的用户，密码要保持一致，否则会认证失败
    上述配置中，创建了三个用户，分别为admin、producer和consumer（创建多少个用户，可根据业务需要配置，用户名和密码可自定义设置）

2.4 修改启动脚本
vim /data/kafka/bin/kafka-server-start.sh ，找到 export KAFKA_HEAP_OPTS , 添加jvm 参数为kafka_server_jaas.conf文件：

1. -Djava.security.auth.login.config=/data/kafka/kafka_server_jaas.conf
   

复制代码

2.5 启动kafka服务

1. # 先启动zookeeper
   
2. /data/kafka/bin/zookeeper-server-start.sh -daemon /data/kafka/config/zookeeper.properties  
   
3. 
   
4. # 启动kafka
   
5. /data/kafka/bin/kafka-server-start.sh -daemon /data/kafka/config/server.properties
   

复制代码

3 kafka客户端配置登录认证
3.1 安装kafka客户端

在/root 路径下新拷贝一个kafka安装包，解压后重命名，作为kafka客户端，客户端路径 /root/kafka
3.2 创建客户端认证配置文件

在 /root/kafka 路径下创建一个 kafka_client_jaas.conf 文件：
vim /root/kafka/kafka_client_jaas.conf:

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

复制代码

说明： 这里配置用户名和密码需要和服务端配置的账号密码保持一致，这里配置了producer这个用户
3.3 添加kafka-console-producer.sh认证文件路径，后面启动生产者测试时使用
vim /root/kafka/bin/kafka-console-producer.sh ，找到 “x$KAFKA_HEAP_OPTS”，添加以下参数：

1. -Djava.security.auth.login.config=/root/kafka/kafka_client_jaas.conf
   

复制代码

3.4 添加kafka-console-consumer.sh认证文件路径，后面启动消费者测试时使用
vim /root/kafka/bin/kafka-console-consumer.sh ，找到 “x$KAFKA_HEAP_OPTS”，添加以下参数：

1. -Djava.security.auth.login.config=/root/kafka/kafka_client_jaas.conf
   

复制代码

4 用客户端连接

1. # 开启生产者：
   
2. /root/kafka/bin/kafka-console-producer.sh --broker-list 10.7.2.201:9092 --topic test --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN
   
3. 
   
4. 
   
5. # 开启消费者：
   
6. /root/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.7.2.201:9092 --topic test --from-beginning --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN
   

复制代码

生产者可正常生产数据，消费者能消费到数据
5 python连接kafka（python2环境）
5.1 安装kafka-python模块包

下载模块包：kafka-python-2.0.2.tar.gz，使用手册：https://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.html
解压后，执行以下命令进行安装：

1. python setup.py install
   

复制代码

5.2 向kafka中写入数据

1. # -*- coding: utf-8 -*-
   
2. from __future__ import absolute_import
   
3. from __future__ import print_function
   
4. from kafka import KafkaProducer
   
5. import json
   
6. import sys
   
7. 
   
8. reload(sys)
   
9. sys.setdefaultencoding('utf8')
   
10. 
    
11. def produceLog(topic_name, log):
    
12.     kafka_producer= KafkaProducer(
    
13.                                   sasl_mechanism="PLAIN",
    
14.                                   security_protocol='SASL_PLAINTEXT',
    
15.                                   sasl_plain_username="producer",
    
16.                                   sasl_plain_password="producerpwd",
    
17.                                   bootstrap_servers='10.7.2.201:9092'
    
18.                                   )                                            
    
19.     kafka_producer.send(topic_name, log, partition=0)
    
20.     kafka_producer.flush()
    
21.     kafka_producer.close()
    
22.    
    
23. for i in range(10):
    
24.     sendlog_dict = {"name":"测试"}
    
25.     sendlog_dict['ID'] = i
    
26.     sendlog_srt = json.dumps(sendlog_dict, ensure_ascii=False)
    
27.     produceLog("pythontest", sendlog_srt)
    

复制代码

执行后，在kafka中查看 pythontest 主题中的数据，如下图：

数据写入成功

5.3 消费数据

1. # -*- coding: utf-8 -*-
   
2. from __future__ import absolute_import
   
3. from __future__ import print_function
   
4. from kafka import KafkaConsumer
   
5. from kafka import TopicPartition
   
6. import sys
   
7. 
   
8. reload(sys)
   
9. sys.setdefaultencoding('utf8')
   
10. 
    
11. consumer= KafkaConsumer(
    
12.                         'pythontest',  # 消息主题
    
13.                         group_id="test",
    
14.                         client_id="python",
    
15.                         sasl_mechanism="PLAIN",
    
16.                         security_protocol='SASL_PLAINTEXT',
    
17.                         sasl_plain_username="producer",
    
18.                         sasl_plain_password="producerpwd",
    
19.                         bootstrap_servers='10.7.2.201:9092',
    
20.                                     auto_offset_reset='earliest' # 从头开始消费
    
21.                         )                                            
    
22. 
    
23. print("主题的分区信息：{}".format(consumer.partitions_for_topic('pythontest')))
    
24. print("主题列表：{}".format(consumer.topics()))
    
25. print("当前消费者订阅的主题：{}".format(consumer.subscription()))
    
26. print("当前消费者topic分区信息：{}".format(consumer.assignment()))
    
27. print("当前消费者可消费的偏移量：{}".format(consumer.beginning_offsets(consumer.assignment())))
    
28. 
    
29. for msg in consumer:
    
30.     print(msg.value)
    

复制代码