SpringCloud 出站序列化

如果禁用本機(jī)編碼（這是默認(rèn)設(shè)置），則框架將使用用戶設(shè)置的contentType轉(zhuǎn)換消息（否則，將應(yīng)用默認(rèn)的application/json）。在這種情況下，對(duì)于出站序列化，它將忽略出站上設(shè)置的任何SerDe。

這是在出站上設(shè)置contentType的屬性。

spring.cloud.stream.bindings.output.contentType: application/json

這是啟用本機(jī)編碼的屬性。

spring.cloud.stream.bindings.output.nativeEncoding: true

如果在輸出綁定上啟用了本機(jī)編碼（用戶必須如上所述明確地啟用本機(jī)編碼），則框架將在出站上跳過任何形式的自動(dòng)消息轉(zhuǎn)換。在這種情況下，它將切換到用戶設(shè)置的Serde。將使用在實(shí)際輸出綁定上設(shè)置的valueSerde屬性。這是一個(gè)例子。

spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde

如果未設(shè)置此屬性，則它將使用“默認(rèn)” SerDe：spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde。

值得一提的是，Kafka Streams綁定程序不會(huì)在出站上序列化密鑰-它僅依賴于Kafka本身。因此，您必須在綁定上指定keySerde屬性，否則它將默認(rèn)為應(yīng)用程序范圍的公用keySerde。

綁定級(jí)別密鑰serde：

spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde

公用密鑰序列：

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

如果使用分支，則需要使用多個(gè)輸出綁定。例如，

interface KStreamProcessorWithBranches {

    		@Input("input")
    		KStream<?, ?> input();

    		@Output("output1")
    		KStream<?, ?> output1();

    		@Output("output2")
    		KStream<?, ?> output2();

    		@Output("output3")
    		KStream<?, ?> output3();
    	}

如果設(shè)置了nativeEncoding，則可以如下對(duì)各個(gè)輸出綁定設(shè)置不同的SerDe。

spring.cloud.stream.kafka.streams.bindings.output1.producer.valueSerde=IntegerSerde
spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=StringSerde
spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=JsonSerde

然后，如果您具有SendTo這樣的@SendTo（{“ output1”，“ output2”，“ output3”}），則分支中的KStream[]將應(yīng)用上面定義的正確的SerDe對(duì)象。如果未啟用nativeEncoding，則可以如下在輸出綁定上設(shè)置不同的contentType值。在這種情況下，框架將使用適當(dāng)?shù)南⑥D(zhuǎn)換器在將消息發(fā)送到Kafka之前轉(zhuǎn)換消息。

spring.cloud.stream.bindings.output1.contentType: application/json
spring.cloud.stream.bindings.output2.contentType: application/java-serialzied-object
spring.cloud.stream.bindings.output3.contentType: application/octet-stream