如何为 kafka 流使用多个使用相同主题的转换器？答案

【问题标题】：How to use multiple transformers using the same topic for kafka streams?如何为 kafka 流使用多个使用相同主题的转换器？
【发布时间】：2021-01-18 03:59:14
【问题描述】：

我需要使用多个转换器解析 kafka 上的复杂消息。每个转换器解析消息的一部分并通过在消息上填充一些属性来编辑消息。最后，完全解析的消息使用 Kafka 消费者存储在数据库中。目前，我正在这样做：

streamsBuilder.stream(Topic.A, someConsumer)
       \\ filters messages that have unparsed parts of type X
       .filter(filterX)
       \\ transformer that edits the message and produces new Topic.E messages
       .transform(ParseXandProduceE::new)
       .to(Topic.A, someProducer)

streamsBuilder.stream(Topic.A, someConsumer)
       \\ filters messages that have unparsed parts of type Y
       .filter(filterY)
       \\ transformer that edits the message and produces new Topic.F messages
       .transform(ParseYandProduceF::new)
       .to(Topic.A, someProducer)

Transformer 看起来像：

class ParseXandProduceE implements Transformer<...> {
    @Override
    public KeyValue<String, Message> transform (String key, Message message) {
           message.x = parse(message.rawX);
           context.forward(newKey, message.x, Topic.E);
           return KeyValue.pair(key, message);
    }
}

但是，这很麻烦，相同的消息会多次通过这些流。此外，还有一个消费者在数据库中存储topic.A 的消息。当前，消息在每次转换之前和每次转换之后都会存储多次。每条消息都需要存储一次。

以下方法可行，但似乎不利，因为每个过滤器+转换块都可以干净地放在自己单独的类中：

streamsBuilder.stream(Topic.A, someConsumer)
       \\ transformer that filters and edits the message and produces new Topic.E + Topic.F messages
       .transform(someTransformer)
       .to(Topic.B, someProducer)

并让持久化消费者收听Topic.B。

后一种建议的解决方案是可行的方法，还是有其他方法可以达到相同的结果？也许有源和接收器的完整拓扑配置？如果是这样，在这种情况下会是什么样子？

【问题讨论】：

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

使用单个变压器似乎是最简单的解决方案。因为您有两个独立的过滤器，如果您尝试链接各个运算符，程序会变得更加复杂。如果您知道每条消息只会通过一个过滤器，而不会同时通过两个过滤器，您可以使用branch()：

KStream[] subStreams = stream.branch(new Predicates[]{filterX,filterY});

subStream[0].transform(ParseXandProduceE::new)
            .merge(subStream[1].transform(ParseYandProduceF::new)
            .to(...)

请注意，上述解决方案仅在两个转换器都不需要转换消息时才有效（branch() 将每条消息放入第一个匹配谓词的分支中，但从不放入多个分支中）。因此，如果一条消息可以通过两个过滤器，您需要执行这样更复杂的操作：

KStream[] subStreams = stream.branch(new Predicates[]{filterX,filterY});

KStream passedX = subStreams[0];
KStream transformedXE = passedX.transform(ParseXandProduceE::new);

// a message that passed filterX may also pass filterY,
// and thus we merge those message back to the "y-stream"
// (of course, those messages would already be transformed by `ParseXandProduceE`)
KStream passedY = subStream[1].merge(transformedXE.filter(filterY);

// the result contains all message that only pass filterX and got transformed,
// plus all messages that passed filterY (and maybe also filterX) and got transformed
KStream result = transformedXE.filterNot(filterY)
                              .merge(passedY.transform(ParseYandProduceF::new)

result.to(...)

【讨论】：