package com.fzzy.conf;
|
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang.StringUtils;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.boot.context.properties.ConfigurationProperties;
|
import org.springframework.context.annotation.Configuration;
|
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
|
import org.springframework.kafka.core.KafkaTemplate;
|
|
import javax.annotation.PostConstruct;
|
import java.util.HashMap;
|
import java.util.Map;
|
|
/**
|
* 自定义配置类
|
*/
|
@Slf4j
|
@Configuration
|
@ConfigurationProperties(prefix = "spring.kafka")
|
public class MyPartitionTemplate {
|
|
private String bootstrapServers;
|
|
|
private String securityProtocol;
|
|
|
private String saslMechanism;
|
|
|
private String saslJaasConfig;
|
|
|
private String saslUsername;
|
|
|
private String saslPassword;
|
|
KafkaTemplate<String, String> kafkaTemplate;
|
|
@PostConstruct
|
public void setKafkaTemplate() {
|
Map<String, Object> props = new HashMap<>();
|
|
props.put("bootstrap.servers", bootstrapServers);
|
props.put("security.protocol", securityProtocol);
|
props.put("sasl.mechanism", saslMechanism);
|
|
if (StringUtils.isNotEmpty(saslUsername) && StringUtils.isNotEmpty(saslPassword)) {
|
saslJaasConfig = saslJaasConfig.replace("{username}", saslUsername)
|
.replace("{password}", saslPassword);
|
props.put("sasl.jaas.config", saslJaasConfig);
|
|
log.debug("----sasl.jaas.config---{}", saslJaasConfig);
|
}
|
|
this.kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
|
}
|
|
public KafkaTemplate<String, String> getKafkaTemplate() {
|
return kafkaTemplate;
|
}
|
|
}
|
