以下屬性僅適用於 Kafka 消費者，並且必須以 spring.cloud.stream.kafka.bindings.<channelName>.consumer. 為字首。

應用程式啟動時，每個分配的分割槽的初始位置取決於 startOffset 和 resetOffsets 這兩個屬性。如果 resetOffsets 為 false，則應用正常的 Kafka 消費者 auto.offset.reset 語義。即，如果某個分割槽的繫結消費者組沒有已提交的偏移量，則位置為 earliest 或 latest。預設情況下，具有顯式 group 的繫結使用 earliest，而匿名繫結（沒有 group）使用 latest。這些預設值可以透過設定 startOffset 繫結屬性來覆蓋。繫結首次使用特定 group 啟動時，將沒有已提交的偏移量。沒有已提交偏移量的另一種情況是偏移量已過期。對於現代 broker（自 2.1 版本起）以及預設的 broker 屬性，偏移量在最後一個成員離開組後 7 天過期。有關詳細資訊，請參閱 offsets.retention.minutes broker 屬性。

當 resetOffsets 為 true 時，繫結器應用與 broker 上沒有已提交偏移量時類似的語義，就好像此繫結從未消費過該主題一樣；即忽略任何當前的已提交偏移量。

以下是可能使用此屬性的兩個用例。

有關主題偏移量的更多控制，請參見使用 KafkaRebalanceListener；當提供了 listener 時，resetOffsets 不應設定為 true，否則會導致錯誤。 >>>>>>> 7bc90c10…​ GH-1084: 新增 txCommitRecovered 屬性

從 3.0 版本開始，當 spring.cloud.stream.binding.<name>.consumer.batch-mode 設定為 true 時，透過輪詢 Kafka Consumer 收到的所有記錄將以 List<?> 的形式呈現給 listener 方法。否則，方法將一次處理一條記錄。批次的大小由 Kafka 消費者屬性 max.poll.records、fetch.min.bytes、fetch.max.wait.ms 控制；有關詳細資訊，請參閱 Kafka 文件。

請注意，@StreamListener 不支援批處理模式 - 它只適用於較新的函數語言程式設計模型。

以下屬性僅適用於 Kafka 生產者，並且必須以 spring.cloud.stream.kafka.bindings.<channelName>.producer. 為字首。

ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)

失敗的傳送會發送到生產者錯誤通道（如果已配置）；參見錯誤通道。預設值：null

+

本節中，我們將展示前述屬性在特定場景下的使用。

預設情況下，記錄使用與原始記錄相同的分割槽釋出到死信主題。這意味著死信主題必須至少擁有與原始記錄相同數量的分割槽。

要更改此行為，請將 DlqPartitionFunction 實現作為 @Bean 新增到應用程式上下文中。只能存在一個此類 bean。該函式會接收消費者組、失敗的 ConsumerRecord 和異常。例如，如果你總是想路由到分割槽 0，你可以使用

也可以為 DLQ 主題定義自定義名稱。為此，請將 DlqDestinationResolver 的實現作為 @Bean 建立到應用程式上下文中。當 binder 檢測到此類 bean 時，它會優先使用，否則將使用 dlqName 屬性。如果兩者都找不到，它將預設為 error.<destination>.<group>。這裡是一個將 DlqDestinationResolver 作為 @Bean 的示例。

提供 DlqDestinationResolver 的實現時，要記住一個重要的事情是，binder 中的 provisioner 不會自動為應用程式建立主題。這是因為 binder 無法推斷出實現可能傳送到的所有 DLQ 主題的名稱。因此，如果你使用此策略提供 DLQ 名稱，確保這些主題事先建立是應用程式的責任。

由於框架無法預料使用者希望如何處理死信訊息，因此它不提供任何標準機制來處理它們。如果死信的原因是臨時的，你可能希望將訊息路由回原始主題。但是，如果問題是永久性的，可能會導致無限迴圈。本主題中的示例 Spring Boot 應用程式展示瞭如何將這些訊息路由回原始主題，但在三次嘗試後將其移動到“停車場”主題。該應用程式是另一個 spring-cloud-stream 應用程式，它從死信主題讀取。當 5 秒內沒有收到訊息時，它會終止。

示例假設原始目的地是 so8400out，消費者組是 so8400。

有幾種策略可以考慮

以下程式碼清單顯示了示例應用程式

從 Spring Cloud Stream 3.0.0 版本開始，Kafka Streams binder 允許應用程式使用 Java 8 中提供的函數語言程式設計風格進行設計和開發。這意味著應用程式可以簡潔地表示為型別為 java.util.function.Function 或 java.util.function.Consumer 的 lambda 表示式。

讓我們看一個非常基礎的例子。

雖然簡單，但這是一個完整的獨立 Spring Boot 應用程式，它利用 Kafka Streams 進行流處理。這是一個消費者應用程式，沒有出站繫結，只有一個入站繫結。應用程式消費資料，並簡單地將來自 KStream 鍵和值的資訊記錄到標準輸出。應用程式包含 SpringBootApplication 註解和一個被標記為 Bean 的方法。該 bean 方法的型別是 java.util.function.Consumer，它使用 KStream 進行引數化。然後在實現中，我們返回一個 Consumer 物件，它本質上是一個 lambda 表示式。在 lambda 表示式內部，提供了處理資料的程式碼。

在此應用程式中，有一個型別為 KStream 的單一輸入繫結。binder 為此應用程式建立一個名為 process-in-0 的繫結，即函式 bean 名稱後跟一個破折號字元 (-)，然後是文字 in，再後跟一個破折號和引數的序數位置。你使用此繫結名稱來設定其他屬性，例如目的地。例如，spring.cloud.stream.bindings.process-in-0.destination=my-topic。

一旦構建為 uber-jar（例如 kstream-consumer-app.jar），就可以按如下方式執行上述示例。

這是另一個示例，它是一個具有輸入和輸出繫結的完整 processor。這是經典的詞頻統計示例，應用程式從主題接收資料，然後在 tumbling time-window 中計算每個單詞的出現次數。

這裡同樣，基本主題與之前的示例相同，但這裡我們有兩個輸入。Java 的 BiFunction 支援用於將輸入繫結到期望的目的地。binder 為輸入生成的預設繫結名稱分別是 process-in-0 和 process-in-1。預設輸出繫結是 process-out-0。在此示例中，BiFunction 的第一個引數被繫結為第一個輸入的 KStream，第二個引數被繫結為第二個輸入的 KTable。

一旦構建為 uber-jar（例如 wordcount-processor.jar），就可以按如下方式執行上述示例。

此應用程式將從 Kafka 主題 words 消費訊息，並將計算結果釋出到輸出主題 counts。

Spring Cloud Stream 將確保來自入站和出站主題的訊息自動繫結為 KStream 物件。作為開發人員，你可以專注於程式碼的業務方面，即編寫 processor 中所需的邏輯。由 Kafka Streams 基礎設施所需的 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.*

儘管上面概述的函數語言程式設計模型是首選方法，如果你願意，仍然可以使用經典的基於 StreamListener 的方法。

這裡有一些示例。

以下是使用 StreamListener 實現的詞頻統計示例的等價版本。

如你所見，這有點冗長，因為你需要提供 EnableBinding 以及其他額外的註解，如 StreamListener 和 SendTo，才能構成一個完整的應用程式。EnableBinding 用於指定包含繫結的繫結介面。在此示例中，我們使用了內建的 KafkaStreamsProcessor 繫結介面，它具有以下契約。

Binder 將為輸入 KStream 和輸出 KStream 建立繫結，因為你使用的繫結介面包含這些宣告。

除了函式式風格提供的程式設計模型上的明顯差異之外，這裡需要提及的一個特別之處在於繫結名稱是你指定在繫結介面中的名稱。例如，在上面的應用程式中，由於我們使用了 KafkaStreamsProcessor，繫結名稱是 input 和 output。繫結屬性需要使用這些名稱。例如 spring.cloud.stream.bindings.input.destination、spring.cloud.stream.bindings.output.destination 等。請記住，這與函式式風格根本不同，因為在函式式風格中，binder 會為應用程式生成繫結名稱。這是因為應用程式在使用 EnableBinding 的函式式模型中不提供任何繫結介面。

這是另一個有兩個輸入的 sink 示例。

以下是上面看到的基於 BiFunction 的 processor 的 StreamListener 等價版本。

最後，這裡是具有三個輸入和 curried 函式的應用程式的 StreamListener 等價版本。

你可能會注意到，上面兩個示例更加冗長，因為除了提供 EnableBinding 之外，你還需要編寫自己的自定義繫結介面。使用函式式模型，可以避免所有這些儀式性的細節。

在我們繼續探討 Kafka Streams binder 提供的通用程式設計模型之前，這裡是多個輸出繫結的 StreamListener 版本。

回顧一下，我們已經回顧了使用 Kafka Streams binder 時的各種程式設計模型選擇。

binder 為輸入提供 KStream、KTable 和 GlobalKTable 的繫結能力。KTable 和 GlobalKTable 繫結僅在輸入端可用。Binder 支援 KStream 的輸入和輸出繫結。

Kafka Streams binder 程式設計模型的要點在於，binder 提供了靈活性，你可以選擇完全函數語言程式設計模型，或者使用基於 StreamListener 的命令式方法。

Binder 允許在單個 Spring Cloud Stream 應用程式中包含多個 Kafka Streams processor。你可以擁有如下所示的應用程式。

在這種情況下，binder 將建立 3 個具有不同 application ID 的獨立 Kafka Streams 物件（更多內容見下文）。但是，如果應用程式中有一個以上的 processor，你必須告訴 Spring Cloud Stream 哪些函式需要被啟用。以下是啟用函式的方法。

spring.cloud.stream.function.definition: process;anotherProcess;yetAnotherProcess

如果你希望某些函式不立即啟用，可以將其從列表中移除。

當你在同一個應用程式中有一個單一的 Kafka Streams processor 和其他型別透過不同 binder 處理的 Function bean 時（例如，基於常規 Kafka Message Channel binder 的函式 bean），這也適用。

Application ID 是 Kafka Streams 應用程式必須提供的強制屬性。Spring Cloud Stream Kafka Streams binder 允許你透過多種方式配置此 application ID。

如果應用程式中只有一個 processor 或 StreamListener，則可以使用以下屬性在 binder 級別進行設定

spring.cloud.stream.kafka.streams.binder.applicationId.

為了方便起見，如果你只有一個 processor，還可以使用 spring.application.name 作為屬性來委託設定 application ID。

如果應用程式中有多個 Kafka Streams processor，則需要為每個 processor 設定 application ID。對於函式式模型，可以將其作為屬性附加到每個函式上。

例如，假設你有以下函式。

和

然後，你可以使用以下 binder 級別屬性為每個函式設定 application ID。

spring.cloud.stream.kafka.streams.binder.functions.process.applicationId

和

spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.applicationId

對於 StreamListener，你需要將其設定在 processor 的第一個輸入繫結上。

例如，假設你有以下兩個基於 StreamListener 的 processor。

那麼你必須使用以下繫結屬性來設定其 application ID。

spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId

和

spring.cloud.stream.kafka.streams.bindings.anotherInput.consumer.applicationId

對於基於函式的模型，這種在繫結級別設定 application ID 的方法也適用。然而，如果你使用函式式模型，像我們上面看到的那樣在 binder 級別按函式設定要容易得多。

對於生產環境部署，強烈建議透過配置顯式指定 application ID。如果你正在自動伸縮應用程式，這一點尤其重要，在這種情況下，你需要確保每個例項都使用相同的 application ID 進行部署。

如果應用程式沒有提供 application ID，那麼在這種情況下，binder 會自動為你生成一個靜態 application ID。這在開發場景中很方便，因為它避免了顯式提供 application ID 的需要。透過這種方式生成的 application ID 在應用程式重啟後將保持靜態。對於函式式模型，生成的 application ID 將是函式 bean 名稱後跟文字 applicationID，例如如果函式 bean 名稱是 process，則生成的 ID 是 process-applicationID。對於 StreamListener，生成的 application ID 不使用函式 bean 名稱，而是使用包含類名後跟方法名，再後跟文字 applicationId。

預設情況下，使用函式式風格時，binder 使用上面討論的策略來生成繫結名稱，即 <function-bean-name>-<in>|<out>-[0..n]，例如 process-in-0, process-out-0 等。如果你想覆蓋這些繫結名稱，可以透過指定以下屬性來實現。

spring.cloud.stream.function.bindings.<default binding name>。預設繫結名稱是 binder 生成的原始繫結名稱。

例如，假設你有這個函式。

Binder 將生成名為 process-in-0、process-in-1 和 process-out-0 的繫結。現在，如果你想將它們完全更改為其他名稱，也許是更具領域特定性的繫結名稱，則可以按如下方式進行。

spring.cloud.stream.function.bindings.process-in-0=users

spring.cloud.stream.function.bindings.process-in-0=regions

和

spring.cloud.stream.function.bindings.process-out-0=clicks

之後，必須在這些新的繫結名稱上設定所有繫結級別屬性。

請記住，使用上面描述的函數語言程式設計模型時，在大多數情況下遵循預設繫結名稱是有意義的。你可能仍然想進行這種覆蓋的唯一原因是你擁有大量的配置屬性，並且希望將繫結對映到更具領域友好性的名稱。

執行 Kafka Streams 應用程式時，必須提供 Kafka broker 伺服器資訊。如果你沒有提供此資訊，binder 會預設期望 broker 執行在 localhost:9092。如果不是這種情況，則需要覆蓋此設定。有幾種方法可以做到這一點。

對於 binder 級別屬性，無論你是否使用透過常規 Kafka binder 提供的 broker 屬性 - spring.cloud.stream.kafka.binder.brokers 都沒有關係。Kafka Streams binder 首先會檢查是否設定了 Kafka Streams binder 特定的 broker 屬性（spring.cloud.stream.kafka.streams.binder.brokers），如果找不到，則會查詢 spring.cloud.stream.kafka.binder.brokers。

鍵始終使用原生 Serdes 進行反序列化。

對於值，預設情況下，入站的反序列化由 Kafka 原生執行。請注意，這是與之前版本 Kafka Streams binder 的預設行為的重大改變，之前版本的反序列化是由框架完成的。

Kafka Streams binder 將嘗試透過檢視 java.util.function.Function|Consumer 或 StreamListener 的型別簽名來推斷匹配的 Serde 型別。以下是它匹配 Serdes 的順序。

如果上述策略都不奏效，那麼應用程式必須透過配置提供 `Serde`。這可以透過兩種方式進行配置 - 繫結級別或預設級別。

首先，繫結器將檢視是否在繫結級別提供了 Serde。例如，如果您有以下處理器：

那麼，您可以使用以下方式提供繫結級別的 Serde：

如果您希望將預設的 key/value Serdes 用於入站反序列化，可以在繫結器級別進行設定。

如果您不希望使用 Kafka 提供的原生解碼，可以依賴 Spring Cloud Stream 提供的訊息轉換功能。由於原生解碼是預設設定，因此為了讓 Spring Cloud Stream 反序列化入站值物件，您需要顯式停用原生解碼。

例如，如果您有上面相同的 BiFunction 處理器，那麼可以使用 spring.cloud.stream.bindings.process-in-0.consumer.nativeDecoding: false。您需要單獨為所有輸入停用原生解碼。否則，對於那些您沒有停用的輸入，原生解碼仍將應用。

預設情況下，Spring Cloud Stream 將使用 application/json 作為內容型別，並使用適當的 json 訊息轉換器。您可以透過使用以下屬性和相應的 MessageConverter bean 來使用自定義訊息轉換器。

出站序列化與入站反序列化基本遵循相同的規則。與入站反序列化一樣，Spring Cloud Stream 的一個主要變化是出站序列化現在由 Kafka 原生處理。在繫結器 3.0 版本之前，這是由框架本身完成的。

出站的 key 總是由 Kafka 使用繫結器推斷出的匹配 Serde 進行序列化。如果無法推斷 key 的型別，則需要透過配置指定。

值 Serdes 的推斷使用與入站反序列化相同的規則。首先檢查出站型別是否與應用程式中提供的 Bean 匹配。如果不匹配，則檢查是否與 Kafka 暴露的 Serde（例如 - Integer, Long, Short, Double, Float, byte[], UUID 和 String）匹配。如果仍然不匹配，則回退到 Spring Kafka 專案提供的 JsonSerde，但在此之前會先檢視預設的 Serde 配置是否存在匹配。請記住，所有這些對應用程式都是透明的。如果這些都不奏效，使用者必須透過配置提供要使用的 Serde。

假設您正在使用上面相同的 BiFunction 處理器。那麼您可以按如下方式配置出站 key/value Serdes。

如果 Serde 推斷失敗，並且沒有提供繫結級別的 Serdes，則繫結器回退到 JsonSerde，但會查詢預設 Serdes 是否有匹配。

預設 serdes 的配置方式與上面描述的反序列化部分相同。

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

如果您的應用程式使用分支功能並且有多個輸出繫結，則需要為每個繫結單獨配置。再次強調，如果繫結器能夠推斷 Serde 型別，則無需進行此配置。

如果您不希望使用 Kafka 提供的原生編碼，但希望使用框架提供的訊息轉換，則需要顯式停用原生編碼，因為原生編碼是預設設定。例如，如果您有上面相同的 BiFunction 處理器，那麼可以使用 spring.cloud.stream.bindings.process-out-0.producer.nativeEncoding: false。在分支的情況下，您需要單獨為所有輸出停用原生編碼。否則，對於那些您沒有停用的輸出，原生編碼仍將應用。

當由 Spring Cloud Stream 執行轉換時，預設情況下，它將使用 application/json 作為內容型別，並使用適當的 json 訊息轉換器。您可以透過使用以下屬性和相應的 MessageConverter bean 來使用自定義訊息轉換器。

當停用原生編碼/解碼時，繫結器將不會像原生 Serdes 那樣進行任何推斷。應用程式需要顯式提供所有配置選項。因此，通常建議在編寫 Spring Cloud Stream Kafka Streams 應用程式時，保持使用預設的序列化/反序列化選項，並堅持使用 Kafka Streams 提供的原生序列化/反序列化。您必須使用框架提供的訊息轉換功能的唯一場景是您的上游生產者使用了特定的序列化策略。在這種情況下，您需要使用匹配的反序列化策略，因為原生機制可能會失敗。當依賴預設的 Serde 機制時，應用程式必須確保繫結器能夠正確地將入站和出站型別對映到適當的 Serde，否則可能會失敗。

值得一提的是，上面概述的資料序列化/反序列化方法僅適用於處理器的邊緣，即 - 入站和出站。您的業務邏輯仍然可能需要呼叫 Kafka Streams API，這些 API 明確需要 Serde 物件。這些仍然是應用程式的責任，必須由開發人員相應地處理。

Kafka Streams 繫結器允許使用以下屬性指定上述反序列化異常處理器。

或

除了上面兩種反序列化異常處理器外，繫結器還提供了第三種，用於將錯誤記錄（poison pills）傳送到 DLQ（死信佇列）主題。以下是啟用此 DLQ 異常處理器的方法。

設定上述屬性後，所有反序列化錯誤的記錄會自動傳送到 DLQ 主題。

您可以透過以下方式設定 DLQ 訊息釋出的 topic 名稱。

您可以提供一個 DlqDestinationResolver 的實現，這是一個函式式介面。DlqDestinationResolver 將 ConsumerRecord 和異常作為輸入，然後允許指定一個 topic 名稱作為輸出。透過訪問 Kafka ConsumerRecord，可以在 BiFunction 的實現中檢查 header 記錄。

以下是提供 DlqDestinationResolver 實現的示例。

提供 DlqDestinationResolver 的實現時，要記住一個重要的事情是，binder 中的 provisioner 不會自動為應用程式建立主題。這是因為 binder 無法推斷出實現可能傳送到的所有 DLQ 主題的名稱。因此，如果你使用此策略提供 DLQ 名稱，確保這些主題事先建立是應用程式的責任。

如果應用程式中存在 DlqDestinationResolver bean，則該 bean 具有更高的優先順序。如果您不想遵循此方法，而是希望使用配置提供靜態 DLQ 名稱，可以設定以下屬性。

如果設定了此屬性，則錯誤記錄將傳送到 custom-dlq 主題。如果應用程式沒有使用上述任何一種策略，它將建立一個名為 error.<input-topic-name>.<application-id> 的 DLQ 主題。例如，如果您的繫結目標主題是 inputTopic 並且應用程式 ID 是 process-applicationId，則預設的 DLQ 主題是 error.inputTopic.process-applicationId。如果您打算啟用 DLQ，始終建議為每個輸入繫結顯式建立一個 DLQ 主題。

屬性 spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler 適用於整個應用程式。這意味著如果同一應用程式中有多個函式或 StreamListener 方法，此屬性將應用於所有這些方法。但是，如果您在單個處理器中有多個處理器或多個輸入繫結，則可以使用繫結器為每個輸入消費者繫結提供的更細粒度的 DLQ 控制。

如果您有以下處理器：

並且您只想在第一個輸入繫結上啟用 DLQ，而在第二個繫結上啟用 logAndSkip，則可以在消費者級別進行設定，如下所示。

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.deserializationExceptionHandler: sendToDlq spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.deserializationExceptionHandler: logAndSkip

以這種方式設定反序列化異常處理器比在繫結器級別設定具有更高的優先順序。

預設情況下，記錄使用與原始記錄相同的分割槽釋出到死信主題。這意味著死信主題必須至少擁有與原始記錄相同數量的分割槽。

要更改此行為，請將 DlqPartitionFunction 實現作為 `@Bean` 新增到應用程式上下文中。只能存在一個這樣的 bean。該函式提供了消費者組（在大多數情況下與應用程式 ID 相同）、失敗的 ConsumerRecord 和異常。例如，如果您總是希望路由到分割槽 0，可以使用：

在使用 Kafka Streams 繫結器中的異常處理功能時，需要牢記幾點。

與上面描述的反序列化異常處理支援不同，繫結器沒有提供用於處理生產異常的此類一流機制。但是，您仍然可以使用 StreamsBuilderFactoryBean 定製器來配置生產異常處理器，您可以在下面的後續章節中找到更多詳細資訊。

使用以下 API 方法檢索與給定儲存和 key 組合關聯的 KeyQueryMetadata 物件。

使用以下 API 方法檢索與給定儲存和 key 組合關聯的 KakfaStreams 物件。

如上所述，繫結器沒有提供一種一流的方式來註冊全域性狀態儲存作為一項功能。為此，您需要使用定製器。以下是如何做到這一點。

再次強調，如果您有多個處理器，您需要透過過濾掉其他 StreamsBuilderFactoryBean 物件（如上所述使用應用程式 ID）將全域性狀態儲存附加到正確的 StreamsBuilder。

在錯誤處理部分，我們提到繫結器沒有提供一種一流的方式來處理生產異常。儘管如此，您仍然可以使用 StreamsBuilderFacotryBean 定製器來註冊生產異常處理器。見下文。

再一次，如果您有多個處理器，您可能需要適當設定正確的 StreamsBuilderFactoryBean。您也可以使用配置屬性新增此類生產異常處理器（更多內容見下文），但如果您選擇採用程式設計式方法，這是一種選擇。

以下屬性可在繫結器級別使用，並且必須以 spring.cloud.stream.kafka.streams.binder. 為字首。

有關所有可用於 streams 配置的屬性的更多資訊，請參閱 Apache Kafka Streams 文件中的 StreamsConfig JavaDocs。您可以在 StreamsConfig 中設定的所有配置都可以透過此屬性設定。使用此屬性時，它適用於整個應用程式，因為這是繫結器級別的屬性。如果應用程式中有多個處理器，所有處理器都將獲取這些屬性。對於像 application.id 這樣的屬性，這會帶來問題，因此您必須仔細檢查如何使用此繫結器級別的 configuration 屬性對映 StreamsConfig 中的屬性。

以下屬性僅適用於 Kafka Streams 生產者，並且必須以 spring.cloud.stream.kafka.streams.bindings.<binding name>.producer. 為字首。為方便起見，如果存在多個輸出 binding 並且它們都需要一個共同值，則可以使用字首 spring.cloud.stream.kafka.streams.default.producer. 進行配置。

streamPartitionerBeanName：要在消費者端使用的自定義出站分割槽器 bean 名稱。應用可以將自定義的 StreamPartitioner 作為 Spring bean 提供，並將該 bean 的名稱提供給生產者，以代替預設分割槽器。

+ 預設值：請參閱上面關於出站分割槽支援的討論。

以下屬性適用於 Kafka Streams 消費者，並且必須以 spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer. 為字首。為方便起見，如果存在多個輸入 binding 並且它們都需要一個共同值，則可以使用字首 spring.cloud.stream.kafka.streams.default.consumer. 進行配置。

注意：在消費者端使用 resetOffsets 對 Kafka Streams binder 沒有影響。與基於訊息通道的 binder 不同，Kafka Streams binder 不會按需定位到開始或結束。

在 Kafka Streams 中，您可以使用 num.stream.threads 屬性控制處理器可以建立的執行緒數。您可以使用上面在 binder、functions、producer 或 consumer 級別描述的各種 configuration 選項來實現這一點。您也可以使用核心 Spring Cloud Stream 為此目的提供的 concurrency 屬性。使用此屬性時，您需要在消費者端使用它。當函式或 StreamListener 中有多個輸入 binding 時，請在第一個輸入 binding 上設定此屬性。例如，設定 spring.cloud.stream.bindings.process-in-0.consumer.concurrency 後，binder 將其轉換為 num.stream.threads。如果您有多個處理器，其中一個處理器定義了 binding 級別的併發性，而其他處理器沒有，則沒有定義 binding 級別併發性的處理器將回退到透過 spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads 指定的 binder 範圍屬性。如果此 binder 配置不可用，則應用將使用 Kafka Streams 設定的預設值。

我們建議在使用 Eclipse 時使用 m2eclipse Eclipse 外掛。如果您尚未安裝 m2eclipse，可以從“Eclipse Marketplace”獲取。

遺憾的是，m2e 尚不支援 Maven 3.3，因此將專案匯入 Eclipse 後，您還需要告訴 m2eclipse 使用專案中的 .settings.xml 檔案。如果您不這樣做，可能會看到與專案中 POM 相關的許多不同錯誤。開啟 Eclipse 偏好設定，展開 Maven 偏好設定，然後選擇 User Settings。在 User Settings 欄位中，單擊 Browse 並導航到您匯入的 Spring Cloud 專案，選擇該專案中的 .settings.xml 檔案。點選 Apply 然後點選 OK 儲存偏好設定更改。

如果您不想使用 m2eclipse，可以使用以下命令生成 Eclipse 專案元資料

生成的 Eclipse 專案可以透過從 file 選單中選擇 import existing projects 來匯入。

[[contributing] == 貢獻

Spring Cloud 以非限制性的 Apache 2.0 許可釋出，並遵循非常標準的 Github 開發流程，使用 Github 跟蹤器處理問題並將 pull requests 合併到 master 分支。如果您想貢獻哪怕是微不足道的東西，請不要猶豫，但請遵循以下準則。