Python 官方文档:入门教程 => 点击学习
目录服务端配置 1. config 目录添加kafka_server_jaas.conf 配置文件2. kafka-run-class.sh 添加3. config/ser
内容:
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
passWord="7f8d9dsf789ds7ffsdfdsfu9"
user_admin="7f8d9dsf789ds7ffsdfdsfu9"
user_alice="xjfkddjfdssifds";
};
KAFKA_OPTS="$KAFKA_OPTS -Djava.security.auth.login.config=/home/finance/App/kafka_2.12-2.5.1/config/kafka_server_jaas.conf"
添加
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
listener, advertised.listeners 添加对应SASL_PLAINTEXT 监听器
listeners=SASL_PLAINTEXT://10.193.196.112:9092
advertised.listeners=SASL_PLAINTEXT://10.193.196.112:9092
就是在java程序中将安全参数配置进来
生产者、消费者属性配置加入一下配置
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配置文件中的参数后,再往里面添加新的参数
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;
@Configuration
@EnableKafka
public class KafkaProducerConfig {
@Autowired
private KafkaProperties kafkaProperties;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
Map<String, Object> 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<String, String> kafkaTemplate() {
Map<String, Object> 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));
}
}
以上为个人经验,希望能给大家一个参考,也希望大家多多支持编程网。
--结束END--
本文标题: kafka添加安全验证配置方式
本文链接: https://lsjlt.com/news/170236.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
2024-03-01
2024-03-01
2024-03-01
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0