前面两节分别介绍了如何使用Pulsar Java客户端库和Go客户端库开发Producer、Consumer。 目前主流的消息中间件都不负责消息在从生产者到消费者上下游传递过程中的类型安全性，而由客户端自己负责消息的序列化和反序列操作来保障消息传递的类型安全。 Pulsar也支持这种客户端的方法，生产者将具体类型的消息对象序列化成字节数组发送到Topic，消费者从Topic接收字节数组并反序列化为具体类型的消息对象。 除了由客户端负责消息类型安全性的方法，Pulsar还提供了一种服务端的方法即Pulsar Schema，本节将学习如何使用Pulsar Schema管理消息数据的类型安全性。

Pulsar Schema介绍

Pulsar有一个内置的Schema Registry，允许客户端为每个Topic上传消息数据的Schema。这样Producer和Consumer就可以通知Pulsar要通过Topic传输什么类型的数据。 使用Schema，Pulsar可以强制执行类型安全，确保生产者和消费者保持同步，客户端不需要再手动进行消息的序列化和反序列化，而由Pulsar Schema在后台执行这些操作。

Pulsar Schema是在Topic上存储和执行的，而不是存在命名空间和租户中的。

Pulsar使用SchemaInfo数据结构来定义Pulsar Schema，在Go客户端中对应的是pulsar.SchemaInfo结构体:

// Encapsulates data around the schema definition
type SchemaInfo struct {
	Name       string
	Schema     string
	Type       SchemaType
	Properties map[string]string
}

在Java客户端库中对应的是org.apache.pulsar.common.schema.SchemaInfo接口:

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface SchemaInfo {

    String getName();

    /**
     * The schema data in AVRO JSON format.
     */
    byte[] getSchema();

    /**
     * The type of schema (AVRO, JSON, PROTOBUF, etc..).
     */
    SchemaType getType();

    /**
     * Additional properties of the schema definition (implementation defined).
     */
    Map<String, String> getProperties();

    String getSchemaDefinition();
}

可以看出一个SchemaInfo由以下字段组成:

Pulsar内置了很多Schema类型，可分为两类: 基础类型和复杂类型

• 对于基础类型，SchemaInfo中的schema字段不存储任何数据，type字段指定基础类型名称后就可以明确如何序列化和反序列化数据。
• 对于复杂类型，SchemaInfo中的schema字段需要指定schema具体schema定义。

每个用topic存储的SchemaInfo都有一个版本。版本信息用于管理topic内发生的schema更改。 Pulsar的Topic存储并管理SchemaInfo的版本，使用对应版本的SchemaInfo产生的消息被标记为对应版本，确保消息被消费时可以使用版本检索对应的SchemaInfo，并使用正确的schema来反序列化数据。

Pulsar Schema的基本使用

管理Pulsar Schema有两种方法: 自动管理和手动管理，具体的内容可查阅官方文档https://pulsar.apache.org/docs/zh-CN/schema-manage/。

这里简单演示一下自动管理的场景，即如果某个schema通过了schema兼容性检查，Pulsar Producer就会自动将此 schema 更新为topic默认创建的schema。这里使用Pulsar的Java Client库开发的生产者和消费者来做一个简单的演示。

要传递的消息的数据结构为:

public class Order {
	private String orderID;
	private String orderName;

	public String getOrderID() {
		return orderID;
	}

	public void setOrderID(String orderID) {
		this.orderID = orderID;
	}

	public String getOrderName() {
		return orderName;
	}

	public void setOrderName(String orderName) {
		this.orderName = orderName;
	}
}

生产者代码如下:

try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
	try (Producer<Order> producer = client.newProducer(Schema.JSON(Order.class)).topic("persistent://study/app1/topic-2").create()) {
		Order order = new Order();
		order.setOrderID(1000L);
		order.setOrderName("the order");
		producer.newMessage()
				.value(order)
				.key(order.orderID.toString())
				.property("p1", "v1")
				.property("p2", "v2")
				.send();
	}
}

消费者代码如下:

try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
	Consumer<Order> consumer = client.newConsumer(Schema.JSON(Order.class))
			.topic("persistent://study/app1/topic-2")
			.subscriptionName("sub-1")
			.subscriptionType(SubscriptionType.Exclusive)
			.subscribe();

	while (true) {
		Message<Order> msg = consumer.receive();
		try {
			System.out.println("Message received: " + msg.getValue().getOrderName());
			consumer.acknowledge(msg);
		} catch (Exception e) {
			consumer.negativeAcknowledge(msg);
		}
	}
}

使用上面的生产者和消费者代码，当生产者发送一条消息，消费者消费之后。使用pulsar-admin查看一下Topic persistent://study/app1/topic-2的schema:

./pulsar-admin schemas get persistent://study/app1/topic-2
{
  "version": 0,
  "schemaInfo": {
    "name": "topic-2",
    "schema": {
      "type": "record",
      "name": "Order",
      "namespace": "com.pulsar.showcase.client.PulsarSchemaTest",
      "fields": [
        {
          "name": "orderID",
          "type": [
            "null",
            "long"
          ]
        },
        {
          "name": "orderName",
          "type": [
            "null",
            "string"
          ]
        }
      ]
    },
    "type": "JSON",
    "properties": {
      "__alwaysAllowNull": "true",
      "__jsr310ConversionEnabled": "false"
    }
  }
}

可以看到Topic persistent://study/app1/topic-2上自动创建了schema。

可以进一步测试将消息类Order的orderID字段的类型由Long修改成String，同步修改生产的代码，使其生产订单ID为字符串类型的消息发送到Broker，会发送失败并抛出了下面的因Schema不兼容而验证失败的异常:

org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException: org.apache.avro.SchemaValidationException: Unable to read schema: 
{
  "type" : "record",
  "name" : "Order",
  "namespace" : "com.pulsar.showcase.client.PulsarSchemaTest",
  "fields" : [ {
    "name" : "orderID",
    "type" : [ "null", "string" ],
    "default" : null
  }, {
    "name" : "orderName",
    "type" : [ "null", "string" ],
    "default" : null
  } ]
}
using schema:
{
  "type" : "record",
  "name" : "Order",
  "namespace" : "com.pulsar.showcase.client.PulsarSchemaTest",
  "fields" : [ {
    "name" : "orderID",
    "type" : [ "null", "long" ],
    "default" : null
  }, {
    "name" : "orderName",
    "type" : [ "null", "string" ],
    "default" : null
  } ]
}

Schema的版本演进和兼容性

Schema不会保持不变，它们会随着应用和服务的迭代而不断更新，以满足新的需求。 Pulsar官方文档https://pulsar.apache.org/docs/zh-CN/schema-evolution-compatibility/中详细介绍了Pulsar Schema如何进行版本演进和保持兼容性的相关内容。 包括: Pulsar对Schema演化提供了怎样的支持；Schema兼容性检查策略；已经在各个兼容性策略下升级客户端(Producer, Consumer)的顺序。

参考

• https://pulsar.apache.org/docs/zh-CN/schema-get-started/
• https://pulsar.apache.org/docs/zh-CN/schema-understand/
• https://pulsar.apache.org/docs/zh-CN/schema-evolution-compatibility/
• https://pulsar.apache.org/docs/zh-CN/schema-manage/