1.Kafka Broker端配置
添加kafka_server_jaas.conf 文件在 kafka/config目录下
配置用户名:
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_alice="alice-secret";
};
username、password为Broker之间通讯需要的用户名密码;
user_admin=“admin” 代表用户 admin,密码 admin,user_后跟随用户名,表示存在admin这个用户,user_alice同理。
注:这里user_admin不可以省略掉,因为访问其他Broker需要用admin用户去进行交互。
将配置的kafka_server_jaas.conf完整路径,传递到Kafka启动脚本kafka-server-start.sh中
进入kafka-server-start.sh,
修改:
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
为:
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"
其中:/opt/kafka/config/kafka_server_jaas.conf 为kafka_server_jaas.conf完整路径
修改kafka/config目录下 server.properties
listeners=SASL_PLAINTEXT://****:9092 //注意,原来为:PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://****:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin //超级用户,注意User大小写敏感
allow.everyone.if.no.acl.found=true
启动所有Broker,至此一个带权限的Broker集群搭建完毕,连接这个Broker会报连接超时
此时系统还有些权限问题,因为加了权限校验,很多kafka/bin目录下的shell脚本不可用,
若想使用关于zk功能的shell脚本就需要配置zk和kafka之间的acl;
若想使用生产消费脚本功能需要配置生产消费者acl;
2.生产者配置
在/opt/kafka/config目录下存放一个kafka_client_jaas.conf文件
内容:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin";
};
这里的用户名密码,为访问Broker需要的用户名密码,这里配置为admin,也可以用步骤1配置的alice用户
修改生产者启动脚本:kafka-console-producer.sh
修改:
![](/qrcode.jpg)
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
为:
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf kafka.tools.ConsoleProducer "$@"
其中:/opt/kafka/config/kafka_client_jaas.conf 为kafka_client_jaas.conf 所在路径
修改 kafka/config/目录下 producer.properties (或者创建个producer.conf 都一样) 添加:
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
调用启动脚本:
./kafka-console-producer.sh --broker-list ****:9092 --topic kht1 --producer.config /opt/kafka/config/producer.properties
能成功写入数据,标志这配置成功,否则,仔细检查配置
3.消费者配置
在/opt/kafka/config目录下存放一个kafka_client_jaas.conf文件(文件名、目录可以自定义)
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin";
};
这里的用户名密码,为访问Broker需要的用户名密码,这里配置为admin,也可以用步骤1配置的alice用户
修改kafka-console-consumer.sh
修改:
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
为:
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/Users/huxi/SourceCode/newenv/reader_jaas.conf kafka.tools.ConsoleConsumer "$@"
其中:/opt/kafka/config/kafka_client_jaas.conf 为kafka_client_jaas.conf 所在路径
修改:kafka/config目录下consumer.properties(也可以自己添加个配置文件,启动时候指定)
添加:
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
启动消费者命令:
./kafka-console-consumer.sh --bootstrap-server ****:9092 --topic kht1 --from-beginning --consumer.config /opt/kafka/config/consumer.properties
可以和步骤2的生产脚本配合,能消费到生产的数据,即为配置成功,否则仔细检查配置
使用Java消费Demo:
使用Java客户端消费,注意1、2、3 三点配置
@Configuration
public class KafkaProducersConfig {
public Map<String, Object> producerConfigs() {
System.setProperty("java.security.auth.login.config", "/opt/kafka/config/kafka_client_jaas.conf"); // 1.环境变量添加,需要输入配置文件的路径
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.*.*.*:9092");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 40960);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put("security.protocol", "SASL_PLAINTEXT"); //2.协议需要配置
props.put("sasl.mechanism", "PLAIN"); //3.协议需要配置
return props;
}
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<String, String>(producerFactory());
}
}
配置消费者权限命令
- 假设你要添加一个acl “以允许198.51.100.0和198.51.100.1,Principal为User:Bob和User:Alice对主题是Test-Topic有Read和Write的执行权限”
./kafka-acls.sh --authorizer-properties zookeeper.connect=****:2181/kafka --add --allow-principal User:Bob --allow-host 198.51.100.0 --operation Read --operation Write --topic kht1
- 查看kafka acl列表
./kafka-acls.sh --authorizer-properties zookeeper.connect=****:2181/kafka --list --topic kht1
- 指定消费者组
./kafka-acls.sh --authorizer-properties zookeeper.connect=****:2181/kafka --add --allow-principal User:* --consumer --topic konghaot1 --group Group-1
注:其他指令./kafka-acls.sh -h
4.Kafka和zookeeper交互脚本访问配置
- 在zookeeper安装根目录的conf目录下,创建zk_server_jaas.conf, 文件内容如下:
user_kafka=“kafka-secret”; 创建用户kafka, 密码是"kafka-secret", 该账号用于kafka broker与zookeeper连接的时候的认证 - 修改zoo.cfg, 添加一下内容
因为认证的时候用到包org.apache.kafka.common.security.plain.PlainLoginModule, 这个是kafka-client.jar里面,所有需要将相应的jar拷贝到 zookeeper安装根目录的lib目录下, 大概要copy这些jar
- 然后就是修改zk的启动参数,
修改 bin/zkEnv.sh, 在文件尾加上
SERVER_JVMFLAGS=" -Djava.security.auth.login.config=$ZOOCFGDIR/zk_server_jaas.conf" - Kafka Broker配置kafka_server_jaas.conf 再添加一个配置,Client,添加后如下:
- 重启,zk、Kafka,生效后,可以调用和zk有关的shell脚本命令
./kafka-topics.sh --describe --zookeeper ****:2181/kafka --topic kht1