具有不同键/值的Spring Cloud Stream中的多个输出绑定（又名分支）

Question

我知道Kafka Streams允许基于指定的谓词将数据分发到多个主题，并且Kafka Streams绑定器使用@StreamListener和Functional Binding方法支持这一点。

...
// return type KStream<?, WordCount>[]

Predicate<Object, WordCount> isEnglish = (k, v) -> v.word.equals("english");
Predicate<Object, WordCount> isFrench =  (k, v) -> v.word.equals("french");
Predicate<Object, WordCount> isSpanish = (k, v) -> v.word.equals("spanish");

return input.
    ... 
    branch(isEnglish, isFrench, isSpanish);

我想知道在返回数据之前如何转换其中一个分支的键或值。假设我希望其中一个分支具有不同于其他分支的密钥类型。

Predicate<Key, Value> isOne = (k, v) -> v.important.equals("1");
Predicate<Key, Value> isTwo =  (k, v) -> v.important.equals("2");

KStream<Key, Value>[] branches = input.branch(isOne, isTwo);

KStream<String, Value> one = branches[0].selectKey((k, v) -> v.importantValue);

我想用这两个流创建一个新的kstream<？,value>[]数组，但由于泛型数组创建错误而无法创建。

我知道这是可能的，从下面的文档摘录中可以看出，可以为每个分支的生产者指定不同的键/值serdes。

spring.cloud.stream.kafka.streams.bindings.output1.producer.valueSerde=IntegerSerde
spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=StringSerde
spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=JsonSerde

感谢你的帮助.

Answer 1

一个选择是创建一个边主题。然后，发送到该副主题的与wordcount不同的每条记录。仍然是wordcount的记录保留在分支的主题中。

我基于使用KStream的Spring cloud streams示例创建了这个代码示例。它不起作用是因为这个想法起作用了。我有一个模拟示例，它使用不同的order对象，并向错误主题发送错误的order。

return input -> {
            KStream<?, WordCount> intermediateStream = input
                    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                    .groupBy((key, value) -> value)
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(6)))
                    .count(Materialized.as("WordCounts-1"))
                    .toStream()
                    .map((key, value) -> new KeyValue<>(null,
                            new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));

            // here we return to the SIDE_TOPIC records with JsonSerde
            intermediateStream
                    .filter((k, v) -> `create another filter`)
                    .map((k, v) -> `transform only this stream`)
                    .to(SIDE_TOPIC, Produced.with(CustomSerdes.String(), new JsonSerde(....)));

            // here we keep using the branch serializer.
            intermediateStream.branch(isEnglish, isFrench, isSpanish);
        }

当您使用带有错误或空值的数据时，此用例是一种常见的方法，您希望将这些数据发送到一个附带主题，即：错误主题。然后您仍然可以保存那些事件以便将来分析它们。