Kafka Streams：如何在发布到不同主题之前转换消息答案

【问题标题】：Kafka Streams: How to transform message before publishing to different topicsKafka Streams：如何在发布到不同主题之前转换消息
【发布时间】：2018-07-20 02:43:33
【问题描述】：

主题 A 上的消息是使用 Kafka Streams 读取的。基于一些验证，消息被发布到主题 B 或主题 C。这很容易使用 Kafka Streams 的 branch 方法实现

KStreamBuilder kStreamBuilder = new KStreamBuilder();
KStream<String, String> kStream = kStreamBuilder.stream(Serdes.String(), Serdes.String(), topicA.split(","));

KStream<String, String>[] splitStreams = kStream.branch(
    (key, value) -> process(value), //Process method is expensive
    (key, value) -> true
);

splitStreams[0].to(Serdes.String(), Serdes.String(), topicB);
splitStreams[1].to(Serdes.String(), Serdes.String(), topicC);

问题：根据消息是否通过验证，在将其发布到主题 B 或主题 C 之前对其进行编辑/附加。如何做我做到了吗？

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

对于转换消息，您可以使用以下KStream 方法：

map（用于根据键和值更新消息）
mapValues（用于根据值更新消息）
selectKey（用于根据键和值更新键）

示例：

splitStreams[0].mapValues(value -> transformValueForTopicB(value))
        .to(Serdes.String(), Serdes.String(), topicB);
splitStreams[1].map((key, value) -> transformValueForTopicC(key, value))
        .to(Serdes.String(), Serdes.String(), topicC);

正如我从branch 方法中的process 方法名称中看到的那样，您可能尝试处理值，但您只需要提供Predicate（返回true 或false）而不进行任何处理.

【讨论】：

我想从我的流程方法中返回 ProcessResult 对象。它将有一个布尔值指示真或假。我想根据 ProcessResult 对象转换 stream1 和 stream2。我不想运行我的 process(value) 方法两次，因为它很昂贵。我怎样才能做到这一点？
好的，所以您可以在.branch() 之前使用process 逻辑调用mapValues，这会将您的值转换为ProcessResult 对象。例如，ProcessResult 可能有字段：yourRequiredValue 和一个布尔字段。在branch 方法中，你可以只检查这个布尔字段，然后你需要一个mapValues 将ProcessResult 对象转换为字符串
KStream<String, ProcessResult>[] splitStreams = kStream.mapValues(value -> process(value)) .branch((key, value) -> value.isRelateToTopicB, (key, value) -> true); splitStreams[0].mapValues(value -> value.getProcessedValue()).to(...);