我正在設(shè)置 Kafka 消費(fèi)者配置，但該配置在類路徑上找不到密鑰庫或信任庫：@EnableKafka@Configurationpublic class KafkaConfig { @Value("${kafka.ssl.keystore}") private String keyStorePath; @Value("${kafka.ssl.truststore}") private String trustStorePath; @Bean public ConsumerFactory<String, String> getConsumerFactory() { Map<String, Object> properties = new HashMap<>(); properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"my-bootstrap.mydomain.com:443"); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1"); properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "client1"); properties.put("enable.auto.commit", "true"); properties.put("auto.commit.interval.ms", "500"); properties.put("session.timeout.ms", "30000"); properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStorePath); properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "password"); properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trustStorePath); properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "password"); properties.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "password"); return new DefaultKafkaConsumerFactory<>(properties); }src/main/resources/ssl密鑰庫和信任庫都位于與配置類相同的 Maven 模塊中的目錄中。我在 application.yml 中設(shè)置了占位符，如下所示：kafka: ssl: keystore: classpath:ssl/kafka-keystore.jks truststore: classpath:ssl/kafka-truststore.jks但是，應(yīng)用程序無法啟動(dòng)，并出現(xiàn)以下異常："org.apache.kafka.common.KafkaException: java.io.FileNotFoundException: classpath:ssl/kafka-keystore.jks (No such file or directory)"我的理解是，使用@Value可以使用classpath:前綴來解析類路徑（請(qǐng)參閱此鏈接） https://www.baeldung.com/spring-classpath-file-access此外，該@Value技術(shù)可以很好地解析同一應(yīng)用程序中反應(yīng)式 WebClient 配置的密鑰庫和信任庫。我需要做什么來解析 Kafka 配置的類路徑？我在這里錯(cuò)過了什么嗎？
2 回答
Qyouu
TA貢獻(xiàn)1786條經(jīng)驗(yàn) 獲得超11個(gè)贊
您注入一個(gè)字符串，它將保留“classpath：”在字符串值中并將其作為屬性提供給 DefaultKafkaConsumerFactory，嘗試注入到 spring 資源中，例如：
import org.springframework.core.io.Resource;
@Value("classpath:path/to/file/in/classpath")
Resource resourceFile;
然后你可以訪問該文件，你可以獲得絕對(duì)路徑，如下所示：
resourceFile.getFile().getAbsolutePath()
這個(gè)想法是你可以提供 DefaultKafkaConsumerFactory 的絕對(duì)路徑
但是您也可以嘗試刪除“classpath：”并像當(dāng)前代碼一樣注入為 String ，這可能取決于 DefaultKafkaConsumerFactory 如何處理該屬性。但我不明白為什么上面的絕對(duì)路徑不起作用。
GCT1015
TA貢獻(xiàn)1827條經(jīng)驗(yàn) 獲得超4個(gè)贊
對(duì)于像我這樣使用 Spring Boot 和 Spring Kafka 并且不重寫 DefaultKafkaConsumerFactory 的人（僅使用屬性進(jìn)行配置），您可以實(shí)現(xiàn)一個(gè)BeanPostProcessor類。它提供了兩種方法：
postProcessAfterInitialization和postProcessBeforeInitialization
工廠鉤子允許對(duì)新 bean 實(shí)例進(jìn)行自定義修改 - 例如，檢查標(biāo)記接口或使用代理包裝 bean。通常，通過標(biāo)記接口等填充 Bean 的后處理器將實(shí)現(xiàn) postProcessBeforeInitialization(java.lang.Object, java.lang.String)，而用代理包裝 Bean 的后處理器通常將實(shí)現(xiàn) postProcessAfterInitialization(java.lang.Object) ，java.lang.String）。
我將 Spring Boot 與 Spring Kafka 一起使用，我只想更改本地配置文件。
在我的代碼示例中，我使用它來覆蓋 Kafka Location 屬性，因?yàn)閷?duì)于 SSL，它不會(huì)從類路徑讀取。
這就是代碼：
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClientConfig;
import java.io.IOException;
import java.util.Arrays;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
@Configuration
@RequiredArgsConstructor
public class KafkaConfiguration implements BeanPostProcessor {
? @Value("${spring.kafka.ssl.key-store-location:}")
? private Resource keyStoreResource;
? @Value("${spring.kafka.properties.schema.registry.ssl.truststore.location:}")
? private Resource trustStoreResource;
? private final Environment environment;
? @SneakyThrows
? @Override
? public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
? ? if (bean instanceof KafkaProperties) {
? ? ? KafkaProperties kafkaProperties = (KafkaProperties) bean;
? ? ? if(isLocalProfileActive()) {
? ? ? ? configureStoreLocation(kafkaProperties);
? ? ? }
? ? }
? ? return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
? }
? private boolean isLocalProfileActive() {
? ? return Arrays.stream(environment.getActiveProfiles()).anyMatch(profile -> "local".equals(profile));
? }
? private void configureStoreLocation(KafkaProperties kafkaProperties) throws IOException {
? ? kafkaProperties.getSsl().setKeyStoreLocation(new FileSystemResource(keyStoreResource.getFile().getAbsolutePath()));
? ? kafkaProperties.getProperties().put(SchemaRegistryClientConfig.CLIENT_NAMESPACE + SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStoreResource.getFile().getAbsolutePath());
? ? kafkaProperties.getSsl().setTrustStoreLocation(new FileSystemResource(trustStoreResource.getFile().getAbsolutePath()));
? ? kafkaProperties.getProperties().put(SchemaRegistryClientConfig.CLIENT_NAMESPACE + SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trustStoreResource.getFile().getAbsolutePath());
? }
}
這樣我就可以在我的屬性文件中添加：
spring.kafka.ssl.key-store-location=classpath:mykeystore.jks
代碼將從中獲取絕對(duì)路徑并設(shè)置它。它還可以根據(jù)配置文件進(jìn)行過濾。
值得一提的是，BeanPostProcessor 會(huì)針對(duì)每個(gè)bean 運(yùn)行，因此請(qǐng)確保您過濾了您想要的內(nèi)容。
添加回答
舉報(bào)
0/150
提交
取消