编程模型
使用 Kafka Streams binder 提供的编程模型时，可以选择使用高级的 Streams DSL 或混合使用高级和低级的 Processor-API。当混合使用高级和低级 API 时，通常通过在 KStream 上调用 transform 或 process API 方法来实现。
函数式风格
从 Spring Cloud Stream 3.0.0 开始，Kafka Streams binder 允许应用程序使用 Java 8 中可用的函数式编程风格进行设计和开发。这意味着应用程序可以简洁地表示为 java.util.function.Function 或 java.util.function.Consumer 类型的 lambda 表达式。
让我们来看一个非常基本的示例。
@SpringBootApplication
public class SimpleConsumerApplication {
@Bean
public java.util.function.Consumer<KStream<Object, String>> process() {
return input ->
input.foreach((key, value) -> {
System.out.println("Key: " + key + " Value: " + value);
});
}
}尽管简单，这是一个完整的独立 Spring Boot 应用程序，它利用 Kafka Streams 进行流处理。这是一个消费者应用程序，没有出站绑定，只有一个入站绑定。该应用程序消费数据，并简单地将来自 KStream 键和值的信息记录到标准输出。该应用程序包含 @SpringBootApplication 注解和一个标记为 @Bean 的方法。该 Bean 方法的类型是 java.util.function.Consumer，并使用 KStream 参数化。然后在实现中，我们返回一个本质上是 lambda 表达式的 Consumer 对象。在 lambda 表达式内部，提供了处理数据的代码。
在此应用程序中，有一个类型为 KStream 的单一输入绑定。Binder 为应用程序创建此绑定，名称为 process-in-0，即函数 Bean 名称后跟一个破折号字符 (-)，然后是字面量 in，再跟一个破折号，最后是参数的序数位置。您可以使用此绑定名称设置其他属性，例如目标。例如，spring.cloud.stream.bindings.process-in-0.destination=my-topic。
如果在绑定上未设置 destination 属性，则会创建一个与绑定同名的主题（如果应用程序有足够的权限），或者预期该主题已经存在。 |
构建成一个 uber-jar（例如，kstream-consumer-app.jar）后，您可以按如下方式运行上述示例。
如果应用程序选择使用 Spring 的 @Component 注解定义函数式 Bean，Binder 也支持该模型。上述函数式 Bean 可以重写如下。
@Component(name = "process")
public class SimpleConsumer implements java.util.function.Consumer<KStream<Object, String>> {
@Override
public void accept(KStream<Object, String> input) {
input.foreach((key, value) -> {
System.out.println("Key: " + key + " Value: " + value);
});
}
}java -jar kstream-consumer-app.jar --spring.cloud.stream.bindings.process-in-0.destination=my-topic这是另一个示例，它是一个具有输入和输出绑定的完整处理器。这是经典的单词计数示例，应用程序从主题接收数据，然后在翻滚时间窗口中计算每个单词出现的次数。
@SpringBootApplication
public class WordCountProcessorApplication {
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>> process() {
return input -> input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Serialized.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(5000))
.count(Materialized.as("word-counts-state-store"))
.toStream()
.map((key, value) -> new KeyValue<>(key.key(), 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);
}
}这里同样是一个完整的 Spring Boot 应用程序。与第一个应用程序的不同之处在于，该 Bean 方法的类型是 java.util.function.Function。Function 的第一个参数化类型用于输入 KStream，第二个用于输出。在方法体中，提供了一个 Function 类型的 lambda 表达式作为实现，给出了实际的业务逻辑。与前面讨论的基于 Consumer 的应用程序类似，这里的输入绑定默认命名为 process-in-0。对于输出，绑定名称也自动设置为 process-out-0。
构建成一个 uber-jar（例如，wordcount-processor.jar）后，您可以按如下方式运行上述示例。
java -jar wordcount-processor.jar --spring.cloud.stream.bindings.process-in-0.destination=words --spring.cloud.stream.bindings.process-out-0.destination=counts此应用程序将从 Kafka 主题 words 消费消息，计算结果发布到输出主题 counts。
Spring Cloud Stream 将确保来自入站和出站主题的消息自动绑定为 KStream 对象。作为开发人员，您可以专注于代码的业务方面，即编写处理器中所需的逻辑。Kafka Streams 基础设施所需的特定配置设置由框架自动处理。
我们上面看到的两个示例都有一个 KStream 输入绑定。在这两种情况下，绑定都接收来自单个主题的记录。如果您想将多个主题多路复用到单个 KStream 绑定中，您可以在下面提供逗号分隔的 Kafka 主题作为目标。
spring.cloud.stream.bindings.process-in-0.destination=topic-1,topic-2,topic-3
此外，如果您想使用正则表达式匹配主题，也可以将主题模式作为目标提供。
spring.cloud.stream.bindings.process-in-0.destination=input.*
多个输入绑定
许多非简单的 Kafka Streams 应用程序通常通过多个绑定从一个以上的主题消费数据。例如，一个主题作为 KStream 消费，另一个作为 KTable 或 GlobalKTable 消费。应用程序可能希望将数据接收为表类型的原因有很多。考虑一个用例，其中底层主题通过数据库的变更数据捕获 (CDC) 机制填充，或者应用程序只关心下游处理的最新更新。如果应用程序指定数据需要绑定为 KTable 或 GlobalKTable，则 Kafka Streams binder 会将目标正确绑定到 KTable 或 GlobalKTable，并使其可供应用程序操作。我们将介绍 Kafka Streams binder 中处理多个输入绑定的几种不同场景。
Kafka Streams Binder 中的 BiFunction
这是一个我们有两个输入和一个输出的示例。在这种情况下，应用程序可以利用 java.util.function.BiFunction。
@Bean
public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, Long>> process() {
return (userClicksStream, userRegionsTable) -> (userClicksStream
.leftJoin(userRegionsTable, (clicks, region) -> new RegionWithClicks(region == null ?
"UNKNOWN" : region, clicks),
Joined.with(Serdes.String(), Serdes.Long(), null))
.map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(),
regionWithClicks.getClicks()))
.groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
.reduce(Long::sum)
.toStream());
}这里，基本主题与前面的示例相同，但我们有两个输入。Java 的 BiFunction 支持用于将输入绑定到所需的目标。Binder 为输入生成的默认绑定名称分别为 process-in-0 和 process-in-1。默认的输出绑定是 process-out-0。在此示例中，BiFunction 的第一个参数绑定为第一个输入的 KStream，第二个参数绑定为第二个输入的 KTable。
Kafka Streams Binder 中的 BiConsumer
如果有两个输入但没有输出，则可以使用 java.util.function.BiConsumer，如下所示。
@Bean
public BiConsumer<KStream<String, Long>, KTable<String, String>> process() {
return (userClicksStream, userRegionsTable) -> {}
}两个输入以上
如果您有两个以上的输入怎么办？在某些情况下，您需要两个以上的输入。在这种情况下，Binder 允许您链接部分函数。在函数式编程术语中，这种技术通常称为柯里化 (currying)。随着 Java 8 添加的函数式编程支持，Java 现在使您能够编写柯里化函数。Spring Cloud Stream Kafka Streams binder 可以利用此特性来实现多个输入绑定。
让我们来看一个示例。
@Bean
public Function<KStream<Long, Order>,
Function<GlobalKTable<Long, Customer>,
Function<GlobalKTable<Long, Product>, KStream<Long, EnrichedOrder>>>> enrichOrder() {
return orders -> (
customers -> (
products -> (
orders.join(customers,
(orderId, order) -> order.getCustomerId(),
(order, customer) -> new CustomerOrder(customer, order))
.join(products,
(orderId, customerOrder) -> customerOrder
.productId(),
(customerOrder, product) -> {
EnrichedOrder enrichedOrder = new EnrichedOrder();
enrichedOrder.setProduct(product);
enrichedOrder.setCustomer(customerOrder.customer);
enrichedOrder.setOrder(customerOrder.order);
return enrichedOrder;
})
)
)
);
}让我们看看上面提出的绑定模型的细节。在此模型中，我们在入站端有 3 个部分应用的函数。我们将它们称为 f(x)、f(y) 和 f(z)。如果我们将这些函数按真正的数学函数意义展开，它将看起来像这样：f(x) → (fy) → f(z) → KStream<Long, EnrichedOrder>。变量 x 代表 KStream<Long, Order>，变量 y 代表 GlobalKTable<Long, Customer>，变量 z 代表 GlobalKTable<Long, Product>。第一个函数 f(x) 具有应用程序的第一个输入绑定 (KStream<Long, Order>)，其输出是函数 f(y)。函数 f(y) 具有应用程序的第二个输入绑定 (GlobalKTable<Long, Customer>)，其输出是另一个函数 f(z)。函数 f(z) 的输入是应用程序的第三个输入 (GlobalKTable<Long, Product>)，其输出是 KStream<Long, EnrichedOrder>，这是应用程序的最终输出绑定。来自三个部分函数（分别是 KStream、GlobalKTable、GlobalKTable）的输入在方法体中可供您作为 lambda 表达式的一部分实现业务逻辑时使用。
输入绑定分别命名为 enrichOrder-in-0、enrichOrder-in-1 和 enrichOrder-in-2。输出绑定命名为 enrichOrder-out-0。
使用柯里化函数，您几乎可以有任意数量的输入。但是，请记住，在 Java 中，如果输入数量以及对应的部分应用函数数量超过少量，可能会导致代码难以阅读。因此，如果您的 Kafka Streams 应用程序需要超过合理较少数量的输入绑定，并且您想使用这种函数式模型，那么您可能需要重新考虑您的设计并适当地分解应用程序。
输出绑定
Kafka Streams binder 允许使用 KStream 或 KTable 作为输出绑定类型。在幕后，Binder 使用 KStream 上的 to 方法将结果记录发送到输出主题。如果应用程序在函数中提供 KTable 作为输出，Binder 仍然通过委托给 KStream 的 to 方法来使用这种技术。
例如，下面的两个函数都可以工作
@Bean
public Function<KStream<String, String>, KTable<String, String>> foo() {
return KStream::toTable;
};
}
@Bean
public Function<KTable<String, String>, KStream<String, String>> bar() {
return KTable::toStream;
}多个输出绑定
Kafka Streams 允许将出站数据写入多个主题。此功能在 Kafka Streams 中称为分支 (branching)。使用多个输出绑定时，需要提供 KStream 数组 (KStream[]) 作为出站返回类型。
这是一个示例
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>[]> process() {
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 -> {
final Map<String, KStream<Object, WordCount>> stringKStreamMap = input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.groupBy((key, value) -> value)
.windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
.count(Materialized.as("WordCounts-branch"))
.toStream()
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value,
new Date(key.window().start()), new Date(key.window().end()))))
.split()
.branch(isEnglish)
.branch(isFrench)
.branch(isSpanish)
.noDefaultBranch();
return stringKStreamMap.values().toArray(new KStream[0]);
};
}编程模型保持不变，但出站参数化类型是 KStream[]。上面函数的默认输出绑定名称分别为 process-out-0、process-out-1、process-out-2。Binder 生成三个输出绑定的原因是它检测到返回的 KStream 数组长度为三。请注意，在此示例中，我们提供了 noDefaultBranch()；如果改用 defaultBranch()，则需要一个额外的输出绑定，本质上返回一个长度为四的 KStream 数组。
基于函数的 Kafka Streams 编程风格总结
总而言之，下表显示了在函数式范例中可以使用的各种选项。
| 输入数量 | 输出数量 | 使用的组件 |
|---|---|---|
1 |
0 |
java.util.function.Consumer |
2 |
0 |
java.util.function.BiConsumer |
1 |
1..n |
java.util.function.Function |
2 |
1..n |
java.util.function.BiFunction |
>= 3 |
0..n |
使用柯里化函数 |
-
在此表中，如果有多个输出，类型简单地变为
KStream[]。
Kafka Streams Binder 中的函数组合
Kafka Streams binder 支持线性拓扑的最小形式的函数组合。使用 Java 函数式 API 支持，您可以编写多个函数，然后使用 andThen 方法自行组合它们。例如，假设您有两个函数如下。
@Bean
public Function<KStream<String, String>, KStream<String, String>> foo() {
return input -> input.peek((s, s2) -> {});
}
@Bean
public Function<KStream<String, String>, KStream<String, Long>> bar() {
return input -> input.peek((s, s2) -> {});
}即使 Binder 不支持函数组合，您也可以如下组合这两个函数。
@Bean
public Function<KStream<String, String>, KStream<String, Long>> composed() {
foo().andThen(bar());
}然后您可以提供 spring.cloud.function.definition=foo;bar;composed 形式的定义。有了 Binder 中的函数组合支持，您无需编写这个进行显式函数组合的第三个函数。
您可以简单地这样做
spring.cloud.function.definition=foo|bar您甚至可以这样做
spring.cloud.function.definition=foo|bar;foo;bar此示例中组合函数的默认绑定名称变为 foobar-in-0 和 foobar-out-0。
Kafka Streams Binder 中函数组合的限制
当您有一个 java.util.function.Function bean 时，它可以与另一个函数或多个函数组合。同一个函数 bean 也可以与一个 java.util.function.Consumer 组合。在这种情况下，consumer 是最后一个被组合的组件。一个函数可以与多个函数组合，然后也可以以一个 java.util.function.Consumer bean 结束。
组合类型为 java.util.function.BiFunction 的 Bean 时，BiFunction 必须是定义中的第一个函数。被组合的实体必须是 java.util.function.Function 或 java.util.function.Consumer 类型。换句话说，您不能取一个 BiFunction bean 然后再与另一个 BiFunction 组合。
您不能与 BiConsumer 类型或 Consumer 是第一个组件的定义进行组合。您也不能与输出是数组（用于分支的 KStream[]）的函数进行组合，除非这是定义中的最后一个组件。
函数定义中的第一个 Function 或 BiFunction 也可以使用柯里化形式。例如，以下是可能的。
@Bean
public Function<KStream<String, String>, Function<KTable<String, String>, KStream<String, String>>> curriedFoo() {
return a -> b ->
a.join(b, (value1, value2) -> value1 + value2);
}
@Bean
public Function<KStream<String, String>, KStream<String, String>> bar() {
return input -> input.mapValues(value -> value + "From-anotherFooFunc");
}函数定义可以是 curriedFoo|bar。在幕后，Binder 将为柯里化函数创建两个输入绑定，并根据定义中的最后一个函数创建一个输出绑定。在这种情况下，默认输入绑定将是 curriedFoobar-in-0 和 curriedFoobar-in-1。此示例的默认输出绑定变为 curriedFoobar-out-0。
关于在函数组合中使用 KTable 作为输出的特别说明
假设您有两个函数如下。
@Bean
public Function<KStream<String, String>, KTable<String, String>> foo() {
return KStream::toTable;
};
}
@Bean
public Function<KTable<String, String>, KStream<String, String>> bar() {
return KTable::toStream;
}您可以将它们组合为 foo|bar，但请记住，第二个函数（在此示例中是 bar）必须有一个 KTable 作为输入，因为第一个函数（foo）的输出是 KTable。
