kafka添加安全验证配置方式

目录

服务端配置

1. config 目录添加kafka_server_jaas.conf 配置文件2. kafka-run-class.sh 添加3. config/server.properties

客户端接入改造

完整配置示例

综合考虑性能影响、管理成本、安全等级要求，接入便利程度。 鉴权采用SASL+PLAINTEXT 方式。每个集群会分配统一的访问账号及密码用于客户端访问。

服务端配置


1. config 目录添加kafka_server_jaas.conf 配置文件

内容：

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

复制代码

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. security.inter.broker.protocol=SASL_PLAINTEXT
3. sasl.mechanism.inter.broker.protocol=PLAIN
5. sasl.enabled.mechanisms=PLAIN
7. listener， advertised.listeners 添加对应SASL_PLAINTEXT 监听器
9. listeners=SASL_PLAINTEXT://10.193.196.112:9092
11. advertised.listeners=SASL_PLAINTEXT://10.193.196.112:9092

复制代码

客户端接入改造

就是在java程序中将安全参数配置进来
生产者、消费者属性配置加入一下配置

1. props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice;");
2. props.put("security.protocol", "SASL_PLAINTEXT");
3. props.put("sasl.mechanism", "PLAIN");

复制代码

springboot 高版本估计有支持在properties文件中直接配置，此处没有验证。

完整配置示例

此方法，可以用KafkaProperties 获取properties配置文件中的参数后，再往里面添加新的参数

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.beans.factory.annotation.Value;
3. import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
6. import org.springframework.kafka.annotation.EnableKafka;
7. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
8. import org.springframework.kafka.config.KafkaListenerContainerFactory;
9. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
10. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
11. import org.springframework.kafka.core.KafkaTemplate;
12. import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
14. import java.util.Map;
16. /**
17. * kafka配置
18. */
19. @Configuration
20. @EnableKafka
21. public class KafkaProducerConfig {
22.         @Autowired
23.         private KafkaProperties kafkaProperties;
25.         /**
26.          * 消费者配置
27.          */
28.         @Bean
29.         public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
30.                 ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
31.                 Map<String, Object> props = kafkaProperties.buildConsumerProperties();
32.                 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=user1 password=pass1;");
33.                 props.put("security.protocol", "SASL_PLAINTEXT");
34.                 props.put("sasl.mechanism", "PLAIN");
35.                 factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
36.                 factory.setConcurrency(2);
37.                 factory.getContainerProperties().setPollTimeout(1500);
38.                 return factory;
39.         }
41.         /**
42.          * 生产者配置
43.          */
44.         @Bean
45.         public KafkaTemplate<String, String> kafkaTemplate() {
46.                 Map<String, Object> props = kafkaProperties.buildProducerProperties();
47.                 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=user1 password=pass1;");
48.                 props.put("security.protocol", "SASL_PLAINTEXT");
49.                 props.put("sasl.mechanism", "PLAIN");
50.                 return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
51.         }
52. }

复制代码

以上为个人经验，希望能给大家一个参考，也希望大家多多支持中国红客联盟。