程式設計模型
使用 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。
