- 综合考虑性能影响、管理成本、安全等级要求，接入便利程度。
- 鉴权采用SASL+PLAINTEXT 方式。
- 每个集群会分配统一的访问账号及密码用于客户端访问。
服务端配置
1. config 目录添加kafka_server_jaas.conf 配置文件
内容：
1 2 3 4 5 6 7 | KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="7f8d9dsf789ds7ffsdfdsfu9"user_admin="7f8d9dsf789ds7ffsdfdsfu9"user_alice="xjfkddjfdssifds";}; |
2. kafka-run-class.sh 添加
1 | KAFKA_OPTS="$KAFKA_OPTS -Djava.security.auth.login.config=/home/finance/App/kafka_2.12-2.5.1/config/kafka_server_jaas.conf" |
3. config/server.properties
添加
1 2 3 4 5 6 7 8 9 10 11 | security.inter.broker.protocol=SASL_PLAINTEXTsasl.mechanism.inter.broker.protocol=PLAINsasl.enabled.mechanisms=PLAINlistener， advertised.listeners 添加对应SASL_PLAINTEXT 监听器listeners=SASL_PLAINTEXT://10.193.196.112:9092advertised.listeners=SASL_PLAINTEXT://10.193.196.112:9092 |
客户端接入改造
就是在java程序中将安全参数配置进来
生产者、消费者属性配置加入一下配置
1 2 3 | props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice;");props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism", "PLAIN"); |
springboot 高版本估计有支持在properties文件中直接配置，此处没有验证。
完整配置示例
此方法，可以用KafkaProperties 获取properties配置文件中的参数后，再往里面添加新的参数
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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.boot.autoconfigure.kafka.KafkaProperties;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.annotation.EnableKafka;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.config.KafkaListenerContainerFactory;import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import org.springframework.kafka.core.DefaultKafkaProducerFactory;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import java.util.Map;/** * kafka配置 */@Configuration@EnableKafkapublic class KafkaProducerConfig { @Autowired private KafkaProperties kafkaProperties; /** * 消费者配置 */ @Bean public KafkaListenerContainerFactory> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); Map props = kafkaProperties.buildConsumerProperties(); props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=user1 password=pass1;"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "PLAIN"); factory.setConsumerFactory(new DefaultKafkaConsumerFactory(props)); factory.setConcurrency(2); factory.getContainerProperties().setPollTimeout(1500); return factory; } /** * 生产者配置 */ @Bean public KafkaTemplate kafkaTemplate() { Map props = kafkaProperties.buildProducerProperties(); props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=user1 password=pass1;"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "PLAIN"); return new KafkaTemplate(new DefaultKafkaProducerFactory(props)); }} |
以上为个人经验，希望能给大家一个参考，也希望大家多多支持IT俱乐部。
