混合使用高级DSL与低级Processor API

DeepSeek V3 中英对照 Mixing high level DSL and low level Processor API

Kafka Streams 提供了两种 API 变体。它包含一个更高级别的类 DSL API，允许您链式调用各种操作，这对于许多函数式编程者来说可能很熟悉。Kafka Streams 还提供了对低级 Processor API 的访问。Processor API 虽然功能非常强大，并且能够在更低的级别控制各种操作，但其本质上是命令式的。Spring Cloud Stream 的 Kafka Streams 绑定器允许您使用高级 DSL，或者混合使用 DSL 和 Processor API。混合使用这两种变体为您提供了许多选项，以控制应用程序中的各种用例。应用程序可以使用 transform 或 process 方法 API 调用来访问 Processor API。

下面我们来看一下如何在 Spring Cloud Stream 应用程序中结合使用 DSL 和处理器 API，特别是通过 process API 来实现。

@Bean
public Consumer<KStream<Object, String>> process() {
    return input ->
        input.process(() -> new Processor<Object, String>() {
            @Override
            @SuppressWarnings("unchecked")
            public void init(ProcessorContext context) {
               this.context = context;
            }

            @Override
            public void process(Object key, String value) {
                //business logic
            }

            @Override
            public void close() {

        });
}

以下是一个使用 transform API 的示例。

@Bean
public Consumer<KStream<Object, String>> process() {
    return (input, a) ->
        input.transform(() -> new Transformer<Object, String, KeyValue<Object, String>>() {
            @Override
            public void init(ProcessorContext context) {

            }

            @Override
            public void close() {

            }

            @Override
            public KeyValue<Object, String> transform(Object key, String value) {
                // business logic - return transformed KStream;
            }
        });
}

process API 方法调用是一个终端操作，而 transform API 是非终端的，它返回一个可能经过转换的 KStream，你可以使用 DSL 或处理器 API 继续对其进行进一步处理。