背景

我们最近在做新业务的技术选型，其中涉及到了对消息中间件的选择；结合我们的实际情况希望它能满足以下几个要求：

友好的云原生支持：因为现在的主力语言是 Go，同时在运维上能够足够简单。
官方支持多种语言的 SDK：还有一些 Python、Java 相关的代码需要维护。
最好是有一些方便好用的特性，比如：延时消息、死信队列、多租户等。

当然还有一些水平扩容、吞吐量、低延迟这些特性就不用多说了，几乎所有成熟的消息中间件都能满足这些要求。

基于以上的筛选条件，Pulsar 进入了我们的视野。

作为 Apache 下的顶级项目，以上特性都能很好的支持。

下面我们来它有什么过人之处。

架构

从官方的架构图中可以看出 Pulsar 主要有以下组件组成：

Broker 无状态组件，可以水平扩展，主要用于生产者、消费者连接；与 Kafka 的 broker 类似，但没有数据存储功能，因此扩展更加轻松。
BookKeeper 集群：主要用于数据的持久化存储。
Zookeeper 用于存储 broker 与 BookKeeper 的元数据。

整体一看似乎比 Kafka 所依赖的组件还多，这样确实会提供系统的复杂性；但同样的好处也很明显。

Pulsar 的存储于计算是分离的，当需要扩容时会非常简单，直接新增 broker 即可，没有其他的心智负担。

当存储成为瓶颈时也只需要扩容 BookKeeper，不需要人为的做重平衡，BookKeeper 会自动负载。

同样的操作，Kafka 就要复杂的多了。

特性

多租户

多租户也是一个刚需功能，可以在同一个集群中对不同业务、团队的数据进行隔离。

persistent://core/order/create-order

以这个 topic 名称为例，在 core 这个租户下有一个 order 的 namespace，最终才是 create-order 的 topic 名称。

在实际使用中租户一般是按照业务团队进行划分，namespace 则是当前团队下的不同业务；这样便可以很清晰的对 topic 进行管理。

通常有对比才会有伤害，在没有多租户的消息中间件中是如何处理这类问题的呢：

干脆不分这么细，所有业务线混着用，当团队较小时可能问题不大；一旦业务增加，管理起来会非常麻烦。
自己在 topic 之前做一层抽象，但其实本质上也是在实现多租户。
各个业务团队各自维护自己的集群，这样当然也能解决问题，但运维复杂度自然也就提高了。

以上就很直观的看出多租户的重要性了。

Function 函数计算

Pulsar 还支持轻量级的函数计算，例如需要对某些消息进行数据清洗、转换，然后再发布到另一个 topic 中。

这类需求就可以编写一个简单的函数，Pulsar 提供了 SDK 可以方便的对数据进行处理，最后使用官方工具发布到 broker 中。

在这之前这类简单的需求可能也需要自己处理流处理引擎。

应用

除此之外的上层应用，比如生产者、消费者这类概念与使用大家都差不多。

比如 Pulsar 支持四种消费模式：

Exclusive：独占模式，同时只有一个消费者可以启动并消费数据；通过 SubscriptionName 标明是同一个消费者），适用范围较小。
Failover 故障转移模式：在独占模式基础之上可以同时启动多个 consumer，一旦一个 consumer 挂掉之后其余的可以快速顶上，但也只有一个 consumer 可以消费；部分场景可用。
Shared 共享模式：可以有 N 个消费者同时运行，消息按照 round-robin 轮询投递到每个 consumer 中；当某个 consumer 宕机没有 ack 时，该消息将会被投递给其他消费者。这种消费模式可以提高消费能力，但消息无法做到有序。
KeyShared 共享模式：基于共享模式；相当于对同一个topic中的消息进行分组，同一分组内的消息只能被同一个消费者有序消费。

第三种共享消费模式应该是使用最多的，当对消息有顺序要求时可以使用 KeyShared 模式。

SDK

官方支持的 SDK 非常丰富；我也在官方的 SDK 的基础之上封装了一个内部使用的 SDK。

因为我们使用了 dig 这样的轻量级依赖注入库，所以使用起来大概是这个样子：

 SetUpPulsar(lookupURL)
 container := dig.New()
 container.Provide(func() ConsumerConfigInstance {
  return NewConsumer(&pulsar.ConsumerOptions{
   Topic:            "persistent://core/order/create-order",
   SubscriptionName: "order-sub",
   Type:             pulsar.Shared,
   Name:             "consumer01",
  }, ConsumerOrder)

 })

 container.Provide(func() ConsumerConfigInstance {
  return NewConsumer(&pulsar.ConsumerOptions{
   Topic:            "persistent://core/order/update-order",
   SubscriptionName: "order-sub",
   Type:             pulsar.Shared,
   Name:             "consumer02",
  }, ConsumerInvoice)

 })

 container.Invoke(StartConsumer)