-1

我想在使用Spring Cloud Stream 的 Kafka Binder 的示例应用程序中使用类型为KeyValueStore的 Kafka 状态存储。根据文档，它应该很简单。这是我的主要课程：

@SpringBootApplication
public class KafkaStreamTestApplication {

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

    @Bean
    public BiFunction<KStream<String, String>, KeyValueStore<String,String>, KStream<String, String>> process(){
        return (input,store) -> input.mapValues(v -> v.toUpperCase());

    }


    @Bean
    public StoreBuilder myStore() {
        return Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore("my-store"), Serdes.String(),
                Serdes.String());
    }
}

我想 KeyValueStore 应该作为“process”方法的第二个参数传递，但应用程序无法启动并显示以下消息：

Caused by: java.lang.IllegalStateException: No factory found for binding target type: org.apache.kafka.streams.state.KeyValueStore among registered factories: channelFactory,messageSourceFactory,kStreamBoundElementFactory,kTableBoundElementFactory,globalKTableBoundElementFactory
    at org.springframework.cloud.stream.binding.AbstractBindableProxyFactory.getBindingTargetFactory(AbstractBindableProxyFactory.java:82) ~[spring-cloud-stream-3.0.3.RELEASE.jar:3.0.3.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory.bindInput(KafkaStreamsBindableProxyFactory.java:191) ~[spring-cloud-stream-binder-kafka-streams-3.0.3.RELEASE.jar:3.0.3.RELEASE]
    at org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory.afterPropertiesSet(KafkaStreamsBindableProxyFactory.java:103) ~[spring-cloud-stream-binder-kafka-streams-3.0.3.RELEASE.jar:3.0.3.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1855) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1792) ~[spring-beans-5.2.5.RELEASE.jar:5.2.5.RELEASE]
4

1 回答 1

0

我找到了关于如何使用存储在 Spring Cloud Stream 中读取单元测试的解决方案。

下面的代码是我如何将该解决方案应用于我的代码。转换器使用 Spring bean 方法“ myStore ”提供的 Store

@SpringBootApplication
public class KafkaStreamTestApplication {

    public static final String MY_STORE_NAME = "my-store";

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


    @Bean
    public Function<KStream<String, String>, KStream<String, String>> process2(){
        return (input) -> input.
              transformValues(() -> new MyValueTransformer(), MY_STORE_NAME);

    }


    @Bean
    public StoreBuilder<?> myStore() {
        return Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore(MY_STORE_NAME), Serdes.String(),
                Serdes.String());
    }

}

public class MyValueTransformer implements ValueTransformer<String, String> {

    private KeyValueStore<String,String> store;
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        store = (KeyValueStore<String, String>) this.context.getStateStore(KafkaStreamTestApplication.MY_STORE_NAME);

    }

    @Override
    public String transform(String value) {
        String tValue = store.get(value);

        if(tValue==null) {
            store.put(value, value.toUpperCase());
        }

        return tValue;
    }

    @Override
    public void close() {
        if(store!=null) {
            store.close();
        }
    }

}
于 2020-05-02T14:35:08.937 回答