如何将属性源添加到Spring Cloud Stream Binder的环境中

Question

我编写了一个PropertySource，它为spring启用类路径前缀。Kafka。属性。ssl。信任库。位置（不支持开箱即用）。本质上，这让我可以放置一个信任库。我的Spring Boot应用程序的src/main/resources文件夹中的jks，并从中引用它。jar文件。

这对于纯SpringKafka配置非常有效，如下所示：

spring:
  kafka:
    properties:
      ssl.truststore.location: classpath:myTruststore.jks

当在Spring Cloud Stream活页夹的上下文中给出相同的配置时，它当前会失败：

spring:
  cloud:
    stream:
      binders:
        my-binder:
         type: kafka
         environment:
           spring:
             kafka:
               properties:
                 ssl.truststore.location: classpath:myTruststore.jks

当我期望它以spring.cloud.stream.binders.my-binder.environment.spring.kafka.properties.ssl.truststore.location的poperty名称被调用时，我的Property tySource甚至没有被回调。

我认为我的PropertySource将执行类路径解析，它不是给定Spring Cloud Stream活页夹的环境的一部分。

问题：如何将PropertySources添加到特定绑定器的环境（或所有环境）？

谢啦！

编辑我在Spring Boot自动配置中添加我的属性源，如下所示：

@Configuration
@AutoConfigureBefore(KafkaAutoConfiguration.class)
@ConditionalOnProperty(name = "com.acme.kafka.enabled", matchIfMissing = true)
@EnableConfigurationPropertiesAcmeKafkaConfigurations.class)
public class AcmeKafkaAutoConfiguration {

  @Bean
  ClasspathResourceSupportEnablingPropertySource acmeKafkaClasspathResourceEnablingPropertySource(ConfigurableEnvironment environment) throws IOException {
    ClasspathResourcesSupport classpathResourcesSupport = new ClasspathResourcesSupport(Files.createTempDirectory(ACME_KAFKA_PREFIX));
    ClasspathResourceSupportEnablingPropertySource propertySource 
        = new ClasspathResourceSupportEnablingPropertySource(ClasspathResourceSupportEnablingPropertySource.NAME, environment, classpathResourcesSupport);
    environment.getPropertySources().addFirst(propertySource);
    return propertySource;
  }
}

编辑2：我尝试了Gary Russel下面的建议（使用声明为静态Bean方法的Bean后处理器）。

它可以工作，但在我的情况下，我在表单启动时会收到许多额外的警告日志：

[...]类型的Bean'...'没有资格被所有BeanPostProcess处理（例如：没有资格自动代理）

根据这篇文章，这可能会导致一些非常严重的副作用。

以下是我使用的代码（导致上述警告）：

@Configuration
@AutoConfigureBefore(KafkaAutoConfiguration.class)
@ConditionalOnProperty(name = "com.acme.kafka.enabled", matchIfMissing = true)
@EnableConfigurationProperties(AcmeKafkaConfigurations.class)
public class AcmeKafkaAutoConfiguration {
  private static final String ACME_KAFKA_PREFIX = "acme.kafka.";

  @Bean
  @ConditionalOnMissingBean
  public static List<ConnectivityConfigurationsProvider> acmeKafkaTokenProviders(OAuth2TokenClient oAuthClient, AcmeKafkaConfigurations configuration) {
    List<ConnectivityConfigurationsProvider> connectivityConfigurationsProviders = new ArrayList<>();
    configuration.getInstances().forEach(serviceInstanceConfiguration -> {
      TokenProvider tokenProvider = new DefaultOAuth2TokenProvider(oAuthClient, serviceInstanceConfiguration);
      ConnectivityConfigurationsProvider connectivityConfigurationsProvider = new ConnectivityConfigurationsProvider(serviceInstanceConfiguration, tokenProvider);
      connectivityConfigurationsProviders.add(connectivityConfigurationsProvider);
    });
    return connectivityConfigurationsProviders;
  }

  @Bean
  @ConditionalOnMissingBean
  @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  public static OAuth2TokenClient acmeKafkaOAuth2TokenClient() {
    return new DefaultOAuth2TokenClient(new DefaultClientCredentialsTokenResponseClient());
  }

  @Bean
  public static ConnectivityConfigurationsProviders acmeKafkaConnectivityConfigurationsProviders(AcmeKafkaConfigurations configuration, List<ConnectivityConfigurationsProvider> connectivityConfigurationsProviders) {
    return new ConnectivityConfigurationsProviders(connectivityConfigurationsProviders);
  }

  @Bean
  static NoOpBeanPostProcessor springKafkaConfigurationsPropertySource(ConfigurableEnvironment environment, ConnectivityConfigurationsProviders connectivityConfigurationsProviders) {
    SpringKafkaConfigurationsPropertySource propertySource = new SpringKafkaConfigurationsPropertySource(SpringKafkaConfigurationsPropertySource.NAME, connectivityConfigurationsProviders);
    environment.getPropertySources().addLast(propertySource);
    return new NoOpBeanPostProcessor();
  }

  @Bean
  @ConditionalOnClass(name = "org.springframework.cloud.stream.binder.BinderConfiguration")
  static NoOpBeanPostProcessor springCloudStreamKafkaConfigurationsPropertySource(ConfigurableEnvironment environment, ConnectivityConfigurationsProviders connectivityConfigurationsProviders) {
    SpringCloudStreamKafkaConfigurationsPropertySource propertySource = new SpringCloudStreamKafkaConfigurationsPropertySource(SpringCloudStreamKafkaConfigurationsPropertySource.NAME, connectivityConfigurationsProviders);
    environment.getPropertySources().addLast(propertySource);
    return new NoOpBeanPostProcessor();
  }

  @Bean
  static NoOpBeanPostProcessor acmeKafkaConnectivityConfigurationsPropertySource(ConfigurableEnvironment environment, ConnectivityConfigurationsProviders connectivityConfigurationsProviders) {
    AcmeKafkaConnectivityConfigurationsPropertySource propertySource = new AcmeKafkaConnectivityConfigurationsPropertySource(AcmeKafkaConnectivityConfigurationsPropertySource.NAME, connectivityConfigurationsProviders);
    environment.getPropertySources().addLast(propertySource);
    return new NoOpBeanPostProcessor();
  }

  @Bean
  static NoOpBeanPostProcessor acmeKafkaClasspathResourceEnablingPropertySource(ConfigurableEnvironment environment) throws IOException {
    ClasspathResourcesSupport classpathResourcesSupport = new ClasspathResourcesSupport(Files.createTempDirectory(ACME_KAFKA_PREFIX));
    ClasspathResourceSupportEnablingPropertySource propertySource 
        = new ClasspathResourceSupportEnablingPropertySource(ClasspathResourceSupportEnablingPropertySource.NAME, environment, classpathResourcesSupport);
    environment.getPropertySources().addFirst(propertySource);
    return new NoOpBeanPostProcessor();
  }

  /**
   * This BeanPostProcessor does not really post-process any beans.
   * It is a way of getting the bean methods that add the property sources
   * above to be called early enough in the lifecycle of Spring ApplicationContext
   * creation. 
   * 
   * BeanPostProcessors are instantiated by Spring extremely early.
   * @Bean methods providing them should be declared as static methods.
   * See: https://stackoverflow.com/questions/30874244/bean-annotation-on-a-static-method 
   */
  static class NoOpBeanPostProcessor implements BeanPostProcessor, Ordered {
    @Override
    public int getOrder() {
      return Ordered.LOWEST_PRECEDENCE;
    }
  }
}

Answer 1

编辑

这似乎有效：

@SpringBootApplication
public class So61826877Application {

    private static final String KEY = "spring.cloud.stream.binders.my-binder.environment."
            + "spring.kafka.properties.ssl.truststore.location";

    public static void main(String[] args) {
        SpringApplication.run(So61826877Application.class, args);
    }

    @Bean
    public static BeanPostProcessor configureSource(ConfigurableEnvironment env) {
        Properties source = new Properties();
        System.out.println(env.getProperty(KEY));
        source.setProperty(KEY, "/path/to/myTruststore.jks");
        env.getPropertySources().addFirst(new PropertiesPropertySource("cp", source));
        return new MyBpp();
    }

    @Bean
    Consumer<String> input() {
        return System.out::println;
    }

}

class MyBpp implements BeanPostProcessor, Ordered {

    @Override
    public int getOrder() {
        return Integer.MAX_VALUE;
    }

}

classpath:myTruststore.jks
...
ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 100
    ...
    ssl.truststore.location = /path/to/myTruststore.jks
    ...