快捷導(dǎo)航

kafka添加安全驗(yàn)證配置方式

更新時(shí)間：2022年10月27日 10:01:44 作者：阿拉的夢(mèng)想

這篇文章主要介紹了kafka添加安全驗(yàn)證配置方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

服務(wù)端配置

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

內(nèi)容：

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="7f8d9dsf789ds7ffsdfdsfu9"
user_admin="7f8d9dsf789ds7ffsdfdsfu9"
user_alice="xjfkddjfdssifds";
};

2. kafka-run-class.sh 添加

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

添加

security.inter.broker.protocol=SASL_PLAINTEXT

sasl.mechanism.inter.broker.protocol=PLAIN

sasl.enabled.mechanisms=PLAIN

listener， advertised.listeners 添加對(duì)應(yīng)SASL_PLAINTEXT 監(jiān)聽(tīng)器

listeners=SASL_PLAINTEXT://10.193.196.112:9092

advertised.listeners=SASL_PLAINTEXT://10.193.196.112:9092

客戶端接入改造

就是在java程序中將安全參數(shù)配置進(jìn)來(lái)

生產(chǎn)者、消費(fèi)者屬性配置加入一下配置

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 高版本估計(jì)有支持在properties文件中直接配置，此處沒(méi)有驗(yàn)證。

完整配置示例

此方法，可以用KafkaProperties 獲取properties配置文件中的參數(shù)后，再往里面添加新的參數(shù)

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
@EnableKafka
public class KafkaProducerConfig {
	@Autowired
	private KafkaProperties kafkaProperties;
	
	/**
	 * 消費(fèi)者配置
	 */
	@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;
	}


	/**
	 * 生產(chǎn)者配置
	 */
	@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));
	}
}

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: