MIME類型和Java類型

MIME類型和Java類型

類型轉換Spring Cloud Stream提供的開箱即用如下表所示：“源有效載荷”是指轉換前的有效載荷，“目標有效載荷”是指轉換后的“有效載荷”。類型轉換可以在“生產者”一側（輸出）或“消費者”一側（輸入）上進行。

來源有效載荷	目標有效載荷	content-type標題（來源訊息）	content-type標題（轉換后）	注釋
POJO	JSON String	ignored	application/json
Tuple	JSON String	ignored	application/json	JSON是為Tuple量身定制的
POJO	String (toString())	ignored	text/plain, java.lang.String
POJO	byte[] (java.io serialized)	ignored	application/x-java-serialized-object
JSON byte[] or String	POJO	application/json (or none)	application/x-java-object
byte[] or String	Serializable	application/x-java-serialized-object	application/x-java-object
JSON byte[] or String	Tuple	application/json (or none)	application/x-spring-tuple
byte[]	String	any	text/plain, java.lang.String	將應用在content-type頭中指定的任何Charset
String	byte[]	any	application/octet-stream	將應用在content-type頭中指定的任何Charset

注意	轉換適用於需要類型轉換的有效內容。例如，如果應用程序生成帶有outputType = application / json的XML字符串，則該有效載荷將不會從XML轉換為JSON。這是因為發送到出站通道的有效載荷已經是一個String，所以在運行時不會應用轉換。同樣重要的是要注意，當使用默認的序列化機制時，必須在發送和接收應用程序之間共享有效負載類，並且與二進制內容兼容。當應用程序代碼在兩個應用程序中獨立更改時，這可能會產生問題，因為二進制格式和代碼可能會變得不兼容。

提示	雖然入站和出站渠道都支持轉換，但特別推薦將其用於轉發出站郵件。對於入站郵件的轉換，特別是當目標是POJO時，@StreamListener支持將自動執行轉換。

自定義郵件轉換

除了支持開箱即用的轉換，Spring Cloud Stream還支持注冊您自己的郵件轉換實現。這允許您以各種自定義格式（包括二進制）發送和接收數據，並將其與特定的contentTypes關聯。Spring Cloud Stream將所有類型為org.springframework.messaging.converter.MessageConverter的bean注冊為自定義消息轉換器以及開箱即用消息轉換器。

如果您的消息轉換器需要使用特定的content-type和目標類（用於輸入和輸出），則消息轉換器需要擴展org.springframework.messaging.converter.AbstractMessageConverter。對於使用@StreamListener的轉換，實現org.springframework.messaging.converter.MessageConverter的消息轉換器就足夠了。

以下是在Spring Cloud Stream應用程序中創建消息轉換器bean（內容類型為application/bar）的示例：

@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {

  ...

  @Bean
  public MessageConverter customMessageConverter() {
    return new MyCustomMessageConverter();
  }

public class MyCustomMessageConverter extends AbstractMessageConverter {

	public MyCustomMessageConverter() {
		super(new MimeType("application", "bar"));
	}

	@Override
  protected boolean supports(Class<?> clazz) {
    return (Bar.class == clazz);
  }

	@Override
	protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
		Object payload = message.getPayload();
		return (payload instanceof Bar ? payload : new Bar((byte[]) payload));
	}
}

Spring Cloud Stream還為基於Avro的轉換器和模式演進提供支持。詳情請參閱具體章節。

@StreamListener和訊息轉換

@StreamListener注釋提供了一種方便的方式來轉換傳入的消息，而不需要指定輸入通道的內容類型。在使用@StreamListener注釋的方法的調度過程中，如果參數需要轉換，將自動應用轉換。

例如，讓我們考慮一個帶有{"greeting":"Hello, world"}的String內容的消息，並且在輸入通道上收到application/json的application/json標題。讓我們考慮接收它的以下應用程序：

public class GreetingMessage {

  String greeting;

  public String getGreeting() {
    return greeting;
  }

  public void setGreeting(String greeting) {
    this.greeting = greeting;
  }
}

@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class GreetingSink {

		@StreamListener(Sink.INPUT)
		public void receive(Greeting greeting) {
			// handle Greeting
		}
	}

該方法的參數將自動填充包含JSON字符串的未編組形式的POJO。

Schema進化支持

Spring Cloud Stream通過其spring-cloud-stream-schema模塊為基於模式的消息轉換器提供支持。目前，基於模式的消息轉換器開箱即用的唯一序列化格式是Apache Avro，在將來的版本中可以添加更多的格式。

Apache Avro訊息轉換器

spring-cloud-stream-schema模塊包含可用於Apache Avro序列化的兩種類型的消息轉換器：

• 使用序列化/反序列化對象的類信息的轉換器，或者啟動時已知位置的模式;

• 轉換器使用模式注冊表 - 他們在運行時定位模式，以及隨着域對象的發展動態注冊新模式。

具有模式支持的轉換器

AvroSchemaMessageConverter支持使用預定義模式或使用類中可用的模式信息（反射或包含在SpecificRecord）中的序列化和反序列化消息。如果轉換的目標類型是GenericRecord，則必須設置模式。

對於使用它，您可以簡單地將其添加到應用程序上下文中，可選地指定一個或多個MimeTypes將其關聯。默認MimeType為application/avro。

以下是在注冊Apache Avro MessageConverter的宿應用程序中進行配置的示例，而不需要預定義的模式：

@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {

  ...

  @Bean
  public MessageConverter userMessageConverter() {
      return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
  }
}

相反，這里是一個應用程序，注冊一個具有預定義模式的轉換器，可以在類路徑中找到：

@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {

  ...

  @Bean
  public MessageConverter userMessageConverter() {
      AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
      converter.setSchemaLocation(new ClassPathResource("schemas/User.avro"));
      return converter;
  }
}

為了了解模式注冊表客戶端轉換器，我們將首先描述模式注冊表支持。

Schema注冊表支持

大多數序列化模型，特別是旨在跨不同平台和語言進行可移植性的序列化模型，依賴於描述數據如何在二進制有效載荷中被序列化的模式。為了序列化數據然后解釋它，發送方和接收方都必須訪問描述二進制格式的模式。在某些情況下，可以從序列化的有效載荷類型或從反序列化時的目標類型中推斷出模式，但是在許多情況下，應用程序可以從訪問描述二進制數據格式的顯式模式中受益。模式注冊表允許您以文本格式（通常為JSON）存儲模式信息，並使該信息可訪問需要它的各種應用程序以二進制格式接收和發送數據。一個模式可以作為一個元組引用，它由

• 作為模式的邏輯名稱的主題 ;

• 模式版本 ;

• 描述數據 的二進制格式的模式格式。

Schema注冊服務器

Spring Cloud Stream提供了模式注冊表服務器實現。為了使用它，您可以簡單地將spring-cloud-stream-schema-server工件添加到項目中，並使用@EnableSchemaRegistryServer注釋，將模式注冊表服務器REST控制器添加到應用程序中。此注釋旨在與Spring Boot Web應用程序一起使用，服務器的監聽端口由server.port設置控制。spring.cloud.stream.schema.server.path設置可用於控制模式服務器的根路徑（特別是嵌入其他應用程序時）。spring.cloud.stream.schema.server.allowSchemaDeletion布爾設置可以刪除模式。默認情況下，這是禁用的。

模式注冊表服務器使用關系數據庫來存儲模式。默認情況下，它使用一個嵌入式數據庫。您可以使用Spring Boot SQL數據庫和JDBC配置選項自定義模式存儲。

啟用模式注冊表的Spring Boot應用程序如下所示：

@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaRegistryServerApplication {
	public static void main(String[] args) {
		SpringApplication.run(SchemaRegistryServerApplication.class, args);
	}
}

Schema注冊服務器API

Schema注冊服務器API由以下操作組成：

POST /

注冊一個新的架構

接受具有以下字段的JSON有效載荷：

• subject模式主題;

• format模式格式;

• definition模式定義。

響應是JSON格式的模式對象，包含以下字段：

• id模式標識;

• subject模式主題;

• format模式格式;

• version模式版本;

• definition模式定義。

GET /{subject}/{format}/{version}

根據其主題，格式和版本檢索現有模式。

響應是JSON格式的模式對象，包含以下字段：

• id模式標識;

• subject模式主題;

• format模式格式;

• version模式版本;

• definition模式定義。

GET /{subject}/{format}

根據其主題和格式檢索現有模式的列表。

響應是JSON格式的每個模式對象的模式列表，包含以下字段：

• id模式標識;

• subject模式主題;

• format模式格式;

• version模式版本;

• definition模式定義。

GET /schemas/{id}

通過其id來檢索現有的模式。

響應是JSON格式的模式對象，包含以下字段：

• id模式標識;

• subject模式主題;

• format模式格式;

• version模式版本;

• definition模式定義。

DELETE /{subject}/{format}/{version}

按其主題，格式和版本刪除現有模式。

DELETE /schemas/{id}

按其ID刪除現有模式。

DELETE /{subject}

按其主題刪除現有模式。

注意	本說明僅適用於Spring Cloud Stream 1.1.0.RELEASE的用戶。Spring Cloud Stream 1.1.0.RELEASE使用表名schema存儲Schema對象，這是一些數據庫實現中的關鍵字。為了避免將來發生任何沖突，從1.1.1.RELEASE開始，我們選擇了存儲表的名稱SCHEMA_REPOSITORY。建議任何正在升級的Spring Cloud Stream 1.1.0.RELEASE用戶在升級之前將其現有模式遷移到新表。