Spring Cloud Stream 的 Apache Kafka

Question

这是我的应用程序，它使用来自 Kafka 主题的数据，然后将计算结果发送到一个主题。

@SpringBootApplication
@EnableBinding(KStreamProcessor.class)
public class WordCountProcessorApplication {

    @StreamListener("input")
    @SendTo("output")
    public KStream<?, WordCount> process(KStream<?, String> input) {
        return input
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .groupBy((key, value) -> value)
                .windowedBy(TimeWindows.of(5000))
                .count(Materialized.as("WordCounts-multi"))
                .toStream()
                .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
    }

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

如何在每次使用来自 Kafka 主题的数据之前打印日志？

Answer 1

在拓扑的开头和结尾添加Transformer。

请参阅this discussion，其中有一个请求由框架自动将自定义转换器添加到拓扑中。

决定添加您自己的解决方法就足够了。