关于后端:用-Pulsar-开发多人小游戏五什么是-schema为什么要用-schema

37次阅读

共计 2552 个字符，预计需要花费 7 分钟才能阅读完成。

本文是《用 Pulsar 开发多人在线小游戏》的第三篇，配套源码和全副文档参见我的 GitHub 仓库 play-with-pulsar 以及我的文章列表。

我举荐《数据密集型利用零碎设计》这本书的第四章：编码与演变（在线浏览地址）。

编码（encoding）和演变（evolution）是两个不同的概念，我举例说明一下。

编码是把内存中的数据结构转化成某种格局的字节序列，不便传输或存储 。

咱们最常见的编码方式就是 JSON 了，JSON 的益处就是字符串解决起来很简略，且易于人类浏览；但问题是反对的数据类型不够丰盛，且传输效率不高。为了解决 JSON 的这些问题，就呈现了 XML 格局或者 protobuf 这样的二进制编码协定。

确定了编码方式，生产者就能够欢快地把数据编码成对应的格局发送进 Pulsar，消费者只有依照依照同样的协定解码，就能失去原始数据了。

但问题是，咱们开发的程序是在一直变动的，所以数据自身的构造很可能发生变化，这也就是演变的概念 。

比如说一开始我把这个 User 类序列化成 JSON 字符串而后发到 Pulsar 中：

class User {
    String name;
    int id;
}

然而随着业务倒退，我发现用 int 类型曾经无奈示意用户 ID 了，须要把 id 这个字段改为字符串类型：

class User {
    String name;
    String id;
}

这种状况下，生产者仍然能够把这个类序列化成 JSON 而后发到 Pulsar 中，Pulsar 不关怀数据自身的内容，默认把所有数据都视为字节数组，所以会欣然接受生产者发来的音讯，但消费者那边生产数据时必定会出问题。

那你说，我同时改生产者和消费者的代码还不行吗？能够，但仍然存在很多问题，比如说：

1、在简单的业务场景中，数据处理的逻辑能够非常复杂，生产者和消费者可能横跨多个部门，协调老本很高。

2、必须先下线所有消费者，代码降级完之后能力从新上线。否则消费者忽然生产到无奈辨认的新的数据格式，会产生不可预期的谬误。

所以，让零碎可能更加灵便地适应变动，就是 schema 可能给咱们提供的价值 。

有了 schema 来形容数据的格局，咱们就能够设置数据结构的兼容性（compatibility），比方说咱们能够设置数据向后兼容，即新代码能够读取旧格局的数据；或者设置向前兼容，即旧代码能够读取新格局的数据。

schema 相干的官网文档：

https://pulsar.apache.org/doc…

咱们能够设置 schema 的兼容性，官网文档：

https://pulsar.apache.org/doc…

咱们在创立 Pulsar 的生产者 / 消费者时能够指定音讯的 schema：

// 指定生产者的 schema
Consumer<User> consumer = client.newConsumer(Schema.AVRO(User.class))
        .subscriptionType(SubscriptionType.Shared)
        .topic(topicName)
        .subscriptionName(subscriptionName)
        .subscribe();

// 指定消费者的 schema
Producer<User> producer = client.newProducer(Schema.AVRO(User.class))
        .topic(topicName)
        .create();

每个 topic 的 schema 演变信息都会存在 Pulsar 当中，这样当生产者应用新的 schema 时，Pulsar 会判断新的 schema 是否合乎以后的兼容性设置，如果合乎则更新 topic 对应的 schema，否则的话则会回绝生产者发来的音讯，这样消费者就不会收到不合乎预期的数据了。

在咱们的炸弹人游戏中，玩家能够产生很多不同类型的事件，这些事件应该以什么格局发送到 Pulsar 的 topic 中呢？

我的做法是创立了一个 EventMessage 构造，用 Type 字段标识事件的类型：

type EventMessage struct {
    // Event type
    Type   string `json:"type"`
    Name   string `json:"name"`
    Avatar string `json:"avatar"`
    // Comment stores extra data
    Comment string `json:"comment"`
    X       int    `json:"x"`
    Y       int    `json:"y"`
    Alive   bool   `json:"alive"`
    List    []int  `json:"list"`}

而后用 Avro 的形式定义了一个 JSONSchema：

const eventJsonSchemaDef = `
{
  "type": "record",
  "name": "EventMessage",
  "namespace": "game",
  "fields": [
    {
      "name": "Type",
      "type": "string"
    },
    {
      "name": "Name",
      "type": "string"
    },
    {
      "name": "Avatar",
      "type": "string"
    },
    {
      "name": "Comment",
      "type": "string",
      "default": ""
    },
    {
      "name": "X",
      "type": "int"
    },
    {
      "name": "Y",
      "type": "int"
    },
    {
      "name": "Alive",
      "type": "boolean"
    },
    {
      "name": "List",
        "type": {
            "type": "array",
            "items" : {"type":"int"}
        }
    }
  ]
}
`

// player event topicName
producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:           topicName,
    // use schema to confirm the structure of message
    Schema: pulsar.NewJSONSchema(eventJsonSchemaDef, nil),
})

这样一来，就能够通过 Schema 的束缚防止将来可能产生的很多问题。