以下屬性僅適用於 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 時，binder 應用與 broker 上沒有已提交偏移量時類似的語義，就好像此綁定從未從主題消費過一樣；即，任何當前已提交的偏移量都將被忽略。

以下是可能使用此功能的兩種用例。

有關主題偏移量的更多控制，請參閱 使用 KafkaRebalanceListener；當提供監聽器時，不應將 resetOffsets 設定為 true，否則會導致錯誤。>>>>>>> 7bc90c10…​ GH-1084: Add txCommitRecovered Property

從 3.0 版開始，當 spring.cloud.stream.binding.<name>.consumer.batch-mode 設定為 true 時，通過輪詢 Kafka Consumer 接收的所有記錄將作為 List<?> 呈現給監聽器方法。否則，將一次調用一個記錄的方法。批次的大小由 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。當綁定器偵測到此類 bean 時，它會優先使用它，否則它將使用 dlqName 屬性。如果兩者都找不到，則預設為 error.<destination>.<group>。以下是作為 @Bean 的 DlqDestinationResolver 範例。

在為 DlqDestinationResolver 提供實作時，需要記住的一件重要事情是，綁定器中的佈建器不會自動為應用程式建立主題。這是因為綁定器無法推斷實作可能傳送到的所有 DLQ 主題的名稱。因此，如果您使用此策略提供 DLQ 名稱，則應用程式有責任確保事先建立這些主題。

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

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

有幾種策略需要考慮

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

從 Spring Cloud Stream 3.0.0 開始，Kafka Streams 綁定器允許使用 Java 8 中提供的函數式程式設計風格來設計和開發應用程式。這表示應用程式可以簡潔地表示為 java.util.function.Function 或 java.util.function.Consumer 類型的 Lambda 運算式。

讓我們來看一個非常基本的範例。

雖然很簡單，但這是一個完整的獨立 Spring Boot 應用程式，它利用 Kafka Streams 進行串流處理。這是一個沒有輸出綁定且只有單個輸入綁定的消費者應用程式。該應用程式消費資料，並且僅將來自 KStream 索引鍵和值的資訊記錄在標準輸出上。該應用程式包含 SpringBootApplication 註解和一個標記為 Bean 的方法。bean 方法的類型為 java.util.function.Consumer，它使用 KStream 進行參數化。然後在實作中，我們傳回一個本質上是 Lambda 運算式的 Consumer 物件。在 Lambda 運算式內部，提供了用於處理資料的程式碼。

在此應用程式中，有一個類型為 KStream 的單個輸入綁定。綁定器使用名稱 process-in-0 為應用程式建立此綁定，即函數 bean 名稱後跟一個破折號字元 (-) 和字面值 in，後跟另一個破折號，然後是參數的序數位置。您可以使用此綁定名稱來設定其他屬性，例如目的地。例如，spring.cloud.stream.bindings.process-in-0.destination=my-topic。

一旦建置為 uber-jar（例如，kstream-consumer-app.jar），您可以像下面這樣運行上面的範例。

這是另一個範例，它是一個具有輸入和輸出綁定的完整處理器。這是經典的單字計數範例，其中應用程式從主題接收資料，然後在滾動時間視窗中計算每個單字的出現次數。

同樣，這是一個完整的 Spring Boot 應用程式。此處與第一個應用程式的不同之處在於，bean 方法的類型為 java.util.function.Function。Function 的第一個參數化類型用於輸入 KStream，第二個參數化類型用於輸出。在方法主體中，提供了一個類型為 Function 的 Lambda 運算式，並作為實作，給出了實際的業務邏輯。與先前討論的基於 Consumer 的應用程式類似，此處的輸入綁定預設命名為 process-in-0。對於輸出，綁定名稱也會自動設定為 process-out-0。

一旦建置為 uber-jar（例如，wordcount-processor.jar），您可以像下面這樣運行上面的範例。

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

Spring Cloud Stream 將確保來自輸入和輸出主題的訊息會自動綁定為 KStream 物件。作為開發人員，您可以專注於程式碼的業務方面，即編寫處理器中所需的邏輯。設定 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 綁定介面，它具有以下合約。

綁定器將為輸入 KStream 和輸出 KStream 建立綁定，因為您正在使用包含這些宣告的綁定介面。

除了函數式風格中提供的程式設計模型中的明顯差異之外，此處需要提到的一件特別的事情是綁定名稱是您在綁定介面中指定的名稱。例如，在上面的應用程式中，由於我們正在使用 KafkaStreamsProcessor，因此綁定名稱為 input 和 output。綁定屬性需要使用這些名稱。例如 spring.cloud.stream.bindings.input.destination、spring.cloud.stream.bindings.output.destination 等。請記住，這與函數式風格有根本的不同，因為在函數式風格中，綁定器會為應用程式產生綁定名稱。這是因為應用程式在使用 EnableBinding 的函數式模型中不提供任何綁定介面。

這是另一個接收器的範例，其中我們有兩個輸入。

以下是我們在上面看到的相同基於 BiFunction 的處理器的 StreamListener 等效版本。

最後，這是具有三個輸入和柯里化函數的應用程式的 StreamListener 等效版本。

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

在我們繼續研究 Kafka Streams 綁定器提供的一般程式設計模型之前，這是多個輸出綁定的 StreamListener 版本。

總之，我們回顧了使用 Kafka Streams 綁定器時的各種程式設計模型選擇。

綁定器為輸入端的 KStream、KTable 和 GlobalKTable 提供綁定功能。KTable 和 GlobalKTable 綁定僅在輸入端可用。綁定器支援 KStream 的輸入和輸出綁定。

Kafka Streams 綁定器的程式設計模型的結果是，綁定器為您提供了靈活性，可以選擇完全採用函數式程式設計模型，也可以使用基於 StreamListener 的命令式方法。

綁定器允許在單個 Spring Cloud Stream 應用程式中擁有多個 Kafka Streams 處理器。您可以擁有如下所示的應用程式。

在這種情況下，綁定器將建立 3 個具有不同應用程式 ID 的獨立 Kafka Streams 物件（更多資訊如下）。但是，如果您的應用程式中有一個以上的處理器，則必須告訴 Spring Cloud Stream 哪些函數需要啟動。以下是如何啟動函數。

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

如果您希望某些函數不要立即啟動，您可以從此列表中刪除它們。

當您在同一應用程式中具有單個 Kafka Streams 處理器和其他類型的 Function bean 時，情況也是如此，這些 bean 是透過不同的綁定器處理的（例如，基於常規 Kafka 訊息通道綁定器的函數 bean）

應用程式 ID 是您需要為 Kafka Streams 應用程式提供的強制屬性。Spring Cloud Stream Kafka Streams 綁定器允許您以多種方式配置此應用程式 ID。

如果您的應用程式中只有一個單個處理器或 StreamListener，那麼您可以使用以下屬性在綁定器層級設定此 ID

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

為方便起見，如果您只有一個處理器，您也可以使用 spring.application.name 作為委派應用程式 ID 的屬性。

如果您的應用程式中有多個 Kafka Streams 處理器，則需要為每個處理器設定應用程式 ID。在函數式模型的情況下，您可以將其作為屬性附加到每個函數。

例如，假設您有以下函數。

和

然後，您可以使用以下綁定器層級屬性為每個函數設定應用程式 ID。

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

和

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

在 StreamListener 的情況下，您需要在處理器上的第一個輸入綁定上設定此 ID。

例如，假設您有以下兩個基於 StreamListener 的處理器。

然後，您必須使用以下綁定屬性為此設定應用程式 ID。

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

和

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

對於基於函數的模型，這種在綁定層級設定應用程式 ID 的方法也適用。但是，如果您使用函數式模型，則在綁定器層級為每個函數設定 ID 要容易得多。

對於生產部署，強烈建議透過配置明確指定應用程式 ID。如果您自動擴展應用程式，這種做法尤其重要，在這種情況下，您需要確保使用相同的應用程式 ID 部署每個實例。

如果應用程式未提供應用程式 ID，則在這種情況下，綁定器將為您自動產生靜態應用程式 ID。這在開發情境中很方便，因為它避免了明確提供應用程式 ID 的需要。以此方式產生的應用程式 ID 在應用程式重新啟動時將保持靜態。在函數式模型的情況下，產生的應用程式 ID 將是函數 bean 名稱後跟字面值 applicationID，例如，如果 process 是函數 bean 名稱，則為 process-applicationID。在 StreamListener 的情況下，產生的應用程式 ID 將使用包含類別名稱後跟方法名稱後跟字面值 applicationId，而不是使用函數 bean 名稱。

預設情況下，當使用函數式風格時，綁定器會使用上面討論的策略來產生綁定名稱，即 <function-bean-name>-<in>|<out>-[0..n]，例如 process-in-0、process-out-0 等。如果您想覆寫這些綁定名稱，您可以透過指定以下屬性來完成。

spring.cloud.stream.function.bindings.<default binding name>。預設綁定名稱是綁定器產生的原始綁定名稱。

例如，假設您有這個函數。

綁定器將產生名稱為 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 伺服器資訊。如果您不提供此資訊，則綁定器會預期您在預設 localhost:9092 運行 broker。如果不是這種情況，則需要覆寫它。有幾種方法可以做到這一點。

對於綁定器層級屬性，您是否使用透過常規 Kafka 綁定器提供的 broker 屬性並不重要 - spring.cloud.stream.kafka.binder.brokers。Kafka Streams 綁定器將首先檢查是否設定了 Kafka Streams 綁定器特定的 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。這可以透過兩種方式配置 - 綁定或預設。

首先，binder 將查看是否在綁定層級提供了 Serde。例如，如果您有以下處理器，

那麼，您可以使用以下方式提供綁定層級的 Serde

如果您希望將預設金鑰/值 Serdes 用於入站反序列化，您可以在 binder 層級執行此操作。

如果您不想要 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 原生處理。在 binder 的 3.0 版本之前，這是由框架本身完成的。

出站的金鑰始終由 Kafka 使用 binder 推斷的匹配 Serde 進行序列化。如果它無法推斷金鑰的類型，則需要使用組態指定。

值 serdes 使用與入站反序列化相同的規則進行推斷。首先，它會匹配以查看輸出類型是否來自應用程式中提供的 bean。如果不是，它會檢查是否與 Kafka 公開的 Serde 匹配，例如 - Integer、Long、Short、Double、Float、byte[]、UUID 和 String。如果這不起作用，則它會回退到 Spring Kafka 專案提供的 JsonSerde，但首先查看預設 Serde 組態，以查看是否有匹配項。請記住，所有這些都是對應用程式透明地發生的。如果這些方法皆無效，則使用者必須透過組態提供要使用的 Serde。

假設您正在使用與上述相同的 BiFunction 處理器。那麼您可以將出站金鑰/值 Serdes 配置如下。

如果 Serde 推斷失敗，並且未提供綁定層級的 Serdes，則 binder 會回退到 JsonSerde，但會查看預設 Serdes 是否有匹配項。

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

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

如果您的應用程式使用分支功能並具有多個輸出綁定，則必須針對每個綁定進行配置。再次強調，如果 binder 能夠推斷 Serde 類型，則您不需要進行此配置。

如果您不想要 Kafka 提供的原生編碼，但想要使用框架提供的訊息轉換，則您需要明確停用原生編碼，因為原生編碼是預設值。例如，如果您具有與上述相同的 BiFunction 處理器，則 spring.cloud.stream.bindings.process-out-0.producer.nativeEncoding: false。在分支的情況下，您需要為所有輸出個別停用原生編碼。否則，原生編碼仍將應用於您未停用的那些輸出。

當轉換由 Spring Cloud Stream 完成時，預設情況下，它將使用 application/json 作為內容類型，並使用適當的 json 訊息轉換器。您可以使用以下屬性和對應的 MessageConverter bean 來使用自訂訊息轉換器。

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

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

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

或

除了上述兩個反序列化異常處理器之外，binder 還提供了第三個處理器，用於將錯誤記錄（毒丸）發送到 DLQ（死信佇列）主題。以下是如何啟用此 DLQ 異常處理器。

當設定上述屬性時，所有反序列化錯誤中的記錄都會自動發送到 DLQ 主題。

您可以設定發布 DLQ 訊息的主題名稱，如下所示。

您可以為 DlqDestinationResolver 提供實作，這是一個函數介面。DlqDestinationResolver 接受 ConsumerRecord 和異常作為輸入，然後允許指定主題名稱作為輸出。透過存取 Kafka ConsumerRecord，可以在 BiFunction 的實作中檢查標頭記錄。

以下是為 DlqDestinationResolver 提供實作的範例。

在為 DlqDestinationResolver 提供實作時，需要記住的一件重要事情是，綁定器中的佈建器不會自動為應用程式建立主題。這是因為綁定器無法推斷實作可能傳送到的所有 DLQ 主題的名稱。因此，如果您使用此策略提供 DLQ 名稱，則應用程式有責任確保事先建立這些主題。

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

如果設定了此屬性，則錯誤記錄將發送到主題 custom-dlq。如果應用程式未使用上述任何策略，則它將建立一個 DLQ 主題，名稱為 error.<input-topic-name>.<application-id>。例如，如果您的綁定目的地主題是 inputTopic，且應用程式 ID 是 process-applicationId，則預設的 DLQ 主題是 error.inputTopic.process-applicationId。始終建議為每個輸入綁定明確建立一個 DLQ 主題，如果您的目的是啟用 DLQ。

屬性 spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler 適用於整個應用程式。這表示如果同一個應用程式中有多個函數或 StreamListener 方法，則此屬性將應用於所有這些方法。但是，如果您在單個處理器中有多個處理器或多個輸入綁定，則可以使用 binder 為每個輸入消費者綁定提供的更細緻的 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

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

預設情況下，記錄會使用與原始記錄相同的分割區發布到死信主題。這表示死信主題必須至少具有與原始記錄一樣多的分割區。

若要變更此行為，請將 DlqPartitionFunction 實作新增為應用程式內容中的 @Bean。只能存在一個此類 bean。該函數會收到消費者群組（在大多數情況下與應用程式 ID 相同）、失敗的 ConsumerRecord 和異常。例如，如果您始終想要路由到分割區 0，您可以使用

在使用 Kafka Streams binder 中的異常處理功能時，需要記住幾件事。

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

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

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

如上所述，binder 沒有提供一流的方式來註冊全域狀態儲存區作為一項功能。為此，您需要使用自訂器。以下是如何執行此操作。

同樣，如果您有多個處理器，您希望透過使用如上所述的應用程式 ID 篩選掉其他 StreamsBuilderFactoryBean 物件，將全域狀態儲存區附加到正確的 StreamsBuilder。

在錯誤處理章節中，我們指出 binder 沒有提供一流的方式來處理生產異常。儘管如此，您仍然可以使用 StreamsBuilderFacotryBean 自訂器來註冊生產異常處理器。請參閱下文。

再次強調，如果您有多個處理器，您可能希望針對正確的 StreamsBuilderFactoryBean 適當地設定它。您也可以使用組態屬性新增此類生產異常處理器（有關更多資訊，請參閱下文），但如果您選擇採用程式化方法，這是一個選項。

以下屬性在 binder 層級可用，並且必須以 spring.cloud.stream.kafka.streams.binder. 為前綴。

有關可能進入 streams 配置的所有屬性的更多資訊，請參閱 Apache Kafka Streams 文件中的 StreamsConfig JavaDocs。您可以透過此屬性設定可以從 StreamsConfig 設定的所有配置。當使用此屬性時，它適用於整個應用程式，因為這是一個 binder 層級屬性。如果您的應用程式中有更多處理器，則所有處理器都將獲取這些屬性。在 application.id 等屬性的情況下，這將變得有問題，因此您必須仔細檢查如何使用此 binder 層級 configuration 屬性對應來自 StreamsConfig 的屬性。

以下屬性僅適用於 Kafka Streams 生產者，並且必須以 spring.cloud.stream.kafka.streams.bindings.<binding name>.producer. 為前綴。為了方便起見，如果有多個輸出繫結並且它們都需要一個通用值，則可以使用前綴 spring.cloud.stream.kafka.streams.default.producer. 來配置。

streamPartitionerBeanName：要在消費者端使用的自訂輸出分割器 bean 名稱。應用程式可以提供自訂的 StreamPartitioner 作為 Spring bean，並且可以將此 bean 的名稱提供給生產者以取代預設的分割器。

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

以下屬性適用於 Kafka Streams 消費者，並且必須以 spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer. 為前綴。為了方便起見，如果有多個輸入繫結並且它們都需要一個通用值，則可以使用前綴 spring.cloud.stream.kafka.streams.default.consumer. 來配置。

注意：在消費者端使用 resetOffsets 對 Kafka Streams 繫結器沒有任何影響。與基於訊息通道的繫結器不同，Kafka Streams 繫結器不會按需搜尋開頭或結尾。

在 Kafka Streams 中，您可以使用 num.stream.threads 屬性來控制處理器可以建立的執行緒數量。您可以使用上面在繫結器、函數、生產者或消費者層級下描述的各種 configuration 選項來執行此操作。您也可以使用核心 Spring Cloud Stream 為此目的提供的 concurrency 屬性。使用此屬性時，您需要在消費者端使用它。當您在函數或 StreamListener 中有多個輸入繫結時，請在第一個輸入繫結上設定此屬性。例如，當設定 spring.cloud.stream.bindings.process-in-0.consumer.concurrency 時，繫結器會將其轉換為 num.stream.threads。如果您有多個處理器，並且一個處理器定義了繫結層級並行性，但其他處理器沒有，則那些沒有繫結層級並行性的處理器將預設回通過 spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads 指定的繫結器範圍屬性。如果此繫結器組態不可用，則應用程式將使用 Kafka Streams 設定的預設值。

我們建議在使用 Eclipse 時使用 m2eclipe eclipse 外掛程式。如果您尚未安裝 m2eclipse，可以從“eclipse marketplace”取得。

遺憾的是，m2e 尚不支援 Maven 3.3，因此將專案匯入 Eclipse 後，您還需要告訴 m2eclipse 為專案使用 .settings.xml 檔案。如果您不這樣做，您可能會看到許多與專案中的 POM 相關的不同錯誤。開啟您的 Eclipse 偏好設定，展開 Maven 偏好設定，然後選取使用者設定。在使用者設定欄位中，按一下瀏覽並導航到您匯入的 Spring Cloud 專案，選取該專案中的 .settings.xml 檔案。按一下套用，然後按一下確定以儲存偏好設定變更。

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

可以通過從 file 選單中選取 import existing projects 來匯入產生的 eclipse 專案。

[[contributing] == 貢獻

Spring Cloud 在非限制性的 Apache 2.0 許可下發布，並遵循非常標準的 Github 開發流程，使用 Github 追蹤器來處理問題，並將提取請求合併到 master 中。如果您想貢獻即使是微不足道的事情，也請不要猶豫，但請遵循以下準則。