一、Pulsar 介绍

Apache Pulsar 是 Apache 软件基金会的顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。

Pulsar 的关键特性如下：

• 是下一代云原生分布式消息流平台。
• Pulsar 的单个实例原生支持多个集群，可跨机房在集群间无缝地完成消息复制。
• 极低的发布延迟和端到端延迟。
• 可无缝扩展到超过一百万个 topic。
• 简单的客户端 API，支持 Java、Go、Python 和 C++。
• 主题的多种订阅模式（独占、共享和故障转移）。
• 通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递 。
• 由轻量级的 serverless 计算框架 Pulsar Functions 实现流原生的数据处理。
• 基于 Pulsar Functions 的 serverless connector 框架 Pulsar IO 使得数据更易移入、移出 Apache Pulsar。
• 分层式存储可在数据陈旧时，将数据从热存储卸载到冷/长期存储（如S3、GCS）中。

二、什么是云原生

既然说 Pulsar 是下一代云原生分布式消息流平台，那我们得知道什么是云原生吧。

云原生的概念是 2013 年 Matt Stine 提出的，到目前为止，云原生的概念发生了多次变更，目前最新对云原生的定义为：DevOps + 持续交付 + 微服务 + 容器。

而符合云原生架构的应用程序是：采用开源堆栈（k8s + docker）进行容器化，基于微服务架构提高灵活性和可维护性，借助敏捷方法、DevOps 支持持续迭代和运维自动化，利用云平台设施实现弹性伸缩、动态调度、优化资源利用率。

三、核心概念

3.1 Messages（消息）

Component	Description
Value / data payload	消息携带的数据，所有 Pulsar 的消息携带原始 bytes，但是消息数据也需要遵循数据 schemas。
Key	消息可以被 Key 打标签。这可以对 topic 压缩之类的事情起作用。
Properties	可选的，用户定义属性的 key/value map。
Producer name	生产消息的 producer 的名称（producer 被自动赋予默认名称，但你也可以自己指定。）
Sequence ID	在 topic 中，每个 Pulsar 消息属于一个有序的序列。消息的 sequence ID 是它在序列中的次序。
Publish time	消息发布的时间戳
Event time	可选的时间戳，应用可以附在消息上，代表某个事件发生的时间，例如，消息被处理时。如果没有明确的设置，那么 event time 为0。
TypedMessageBuilder	它用于构造消息。您可以使用TypedMessageBuilder设置消息属性，比如消息键、消息值。设置TypedMessageBuilder时，将键设置为字符串。如果您将键设置为其他类型，例如，AVRO对象，则键将作为字节发送，并且很难从消费者处取回AVRO对象。

消息的默认大小为 5 MB，可以通过以下方式配置消息的最大大小。

• broker.conf

  # The max size of a message (in bytes).
  maxMessageSize=5242880
  

• bookkeeper.conf

  # The max size of the netty frame (in bytes). Any messages received larger than this value are rejected. The default value is 5 MB.
  nettyMaxFrameSizeBytes=5253120
  

3.2 Producers（生产者）

生产者是关联到 topic 的程序，它发布消息到 Pulsar 的 broker 上。

3.2.1 Send modes（发送模式）

producer 可以以同步或者异步的方式发布消息到 broker。

Mode	Description
异步发送	发送消息后，producer等待broker的确认。如果没有收到确认，producer会认为发送失败。
同步发送	producer 将会把消息放入阻塞队列，然后马上返回。客户端类库将会在背后把消息发送给 broker。如果队列满了，根据传给 producer 的参数，producer 可能阻塞或者直接返回失败。

3.2.2 Access mode（访问模式）

你可以为生产者提供不同类型的主题访问模式。

Access mode	Description
Shared（共享）	多个生产者可以发布一个主题，这是默认设置。
Exclusive（独占）	一个主题只能由一个生产者发布。如果已经有生产者连接，其他生产者试图发布该主题立即得到错误。如果“老”生产者与 broker 发生网络分区，“老”生产者将被驱逐，“新”生产者将被选为下一个唯一的生产者。
WaitForExclusive（独占等待）	如果已经有一个生产者连接，生产者的创建是未决的（而不是超时），直到生产者获得独占访问。成功成为排他性的生产者被视为领导者。因此，如果您想为您的应用程序实现 leader 选举方案，您可以使用这种访问模式。

3.2.3 Compression（压缩）

你可以压缩生产者在传输期间发布的消息。Pulsar 目前支持以下类型的压缩：

• LZ4
• ZLIB
• ZSTD
• SNAPPY

3.2.4 Batching（批处理）

如果批处理开启，producer 将会累积一批消息，然后通过一次请求发送出去。批处理的大小取决于最大的消息数量及最大的发布延迟。

3.2.5 Chunking（分块）

• 批处理和分块不能同时启用。要启用分块，必须提前禁用批处理。
• Chunking 只支持持久化的主题。
• Chunking 仅支持 exclusive 和 failover 订阅模式。

3.2.5.1 处理一个 producer 和一个订阅 consumer 的分块消息

如下图所示，当生产者向主题发送一批大的分块消息和普通的非分块消息时。 假设生产者发送的消息为 M1，M1 有三个分块 M1-C1，M1-C2 和 M1-C3。 这个 broker 在其管理的 ledger 里面保存所有的三个块消息，然后以相同的顺序分发给消费者（独占/灾备模式）。 消费者将在内存缓存所有的块消息，直到收到所有的消息块。将这些消息合并成为原始的消息 M1，发送给处理进程。

3.2.5.2 处理多个 producer 和一个订阅 consumer 的分块消息

当多个生产者发布块消息到单个主题，这个 Broker 在同一个 Ledger 里面保存来自不同生产者的所有块消息。 如下所示，生产者1发布的消息 M1，M1 由 M1-C1, M1-C2 和 M1-C3 三个块组成。 生产者2发布的消息 M2，M2 由 M2-C1, M2-C2 和 M2-C3 三个块组成。 这些特定消息的所有分块是顺序排列的，但是其在 ledger 里面可能不是连续的。 这种方式会给消费者带来一定的内存负担。因为消费者会为每个大消息在内存开辟一块缓冲区，以便将所有的块消息合并为原始的大消息。

3.3 Consumers（消费者）

消费者通过订阅关联到主题，然后接收消息的程序。

3.3.1 Receive modes（接收模式）

消息可以通过同步或者异步的方式从 broker 接收。

Mode	Description
同步接收	同步接收将会阻塞，直到消息可用。
异步接收	异步接收立即返回 future 值，例如 java 中的 CompletableFuture，一旦新消息可用，它即刻完成。

3.3.2 Listeners（监听）

客户端类库提供了它们对于 consumer 的监听实现。举一个 Java 客户端的例子，它提供了 MessageListener 接口。在这个接口中，一旦接受到新的消息，received 方法将被调用。

3.3.3 Acknowledgement（确认）

消费者成功处理了消息，需要发送确认给 broker，以让 broker 丢掉这条消息（否则它将存储着此消息）。

消息的确认可以一个接一个，也可以累积一起。累积确认时，消费者只需要确认最后一条它收到的消息。所有之前（包含此条）的消息，都不会被重新发给那个消费者。

累积消息确认不能用于 shared 订阅模式，因为 shared 订阅为同一个订阅引入了多个消费者。

3.4 Topics（主题）

和其它的发布订阅系统一样，Pulsar 中的 topic 是带有名称的通道，用来从 producer 到 consumer 传输消息。Topic 的名称是符合良好结构的 URL。

{persistent|non-persistent}://tenant/namespace/topic

Topic name component	Description
persistent / non-persistent	定义了 topic 类型，Pulsar 支持两种不同 topic：持久和非持久（默认是持久类型，如果你没有指明类型，topic 将会是持久类型）。持久 topic 的所有消息都会保存在硬盘上（这意味着多块硬盘，除非是单机模式的 broker），反之，非持久 topic 的数据不会存储到硬盘上。
tenant	实例中 topic 的租户。tenant 是 Pulsar 多租户的基本要素。可以被跨集群的传播。
namespace	topic 的管理单元，相关 topic 组的管理机制。大多数的 topic 配置在 namespace 层面生效。每个 tenant 可以有多个 namespace。
topic	主题名称的最后组成部分，topic 的名称很自由，没有什么特殊的含义。

3.4.1 Partitioned topics（分区主题）

普通主题仅由单个 broker 提供服务，这限制了主题的最大吞吐量。分区主题是由多个 broker 处理的一种特殊类型的主题，因此允许更高的吞吐量。

分区的主题实际上实现为 N 个内部主题，其中 N 是分区的数量。当将消息发布到分区主题时，每个消息都被路由到几个 broker 中的一个。分区在 broker 间的分布由 Pulsar 自动处理。

如上图，Topic1 主题有 5 个分区（P0 到 P4），划分在 3 个 broker 上。因为分区比 broker 多，前两个 broker 分别处理两个分区，而第三个 broker 只处理一个分区（同样，Pulsar 自动处理分区的分布）。

此主题的消息将广播给两个消费者。路由模式决定将每个消息发布到哪个分区，而订阅模式决定将哪些消息发送到哪个消费者。

在大多数情况下，可以分别决定路由和订阅模式。通常，吞吐量问题应该指导分区/路由决策，而订阅决策应该根据应用程序语义进行指导。

就订阅模式的工作方式而言，分区主题和普通主题之间没有区别，因为分区仅决定消息由生产者发布和由消费者处理和确认之间发生了什么。

分区主题需要通过管理 API 显式创建，分区的数量可以在创建主题时指定。

3.4.1.1 Routing modes（路由模式）

当发布消息到分区 topic，你必须要指定路由模式。路由模式决定了每条消息被发布到的分区（其实是内部主题）。

下面是三种默认可用的路由模式：

Mode	Description
RoundRobinPartition	message 无 key 则轮询，有 key 则 hash(key) 指定分区。（默认模式）
SinglePartition	message 无 key，producer 将会随机选择一个分区，把所有的消息发往该分区。如果 message 指定了 key，分区的 producer 会把 key 做 hash，然后分配消息到指定的分区。
CustomPartition	使用自定义消息路由实现，可以决定特定的消息进入指定的分区。

3.4.1.2 Ordering guarantee（顺序保证）

消息的顺序与路由模式和消息的 key 有关：

Ordering guarantee	Description	Routing Mode and Key
Per-key-partition（按 key 分区）	具有相同 key 的所有消息将被按顺序放置在同一个分区中。	使用 SinglePartition 或 RoundRobinPartition 模式，Key 由每个消息提供。
Per-producer（按 producer）	来自同一生产者的所有消息将是有序的。	使用 SinglePartition 模式，并且没有为每个消息提供 Key。

3.4.1.3 Hashing scheme（哈希方案）

HashingScheme 是一个 enum，表示在选择要为特定消息使用的分区时可用的标准哈希函数集。

有两种类型的标准哈希函数可用：JavaStringHash 和 Murmur3_32Hash。生产者的默认哈希函数是 JavaStringHash。请注意，当生产者可以来自不同的多语言客户端时，JavaStringHash 是没有用的，在这个用例下，建议使用 Murmur3_32Hash。

3.4.2 persistent/Non-persistent topics（持久/非持久主题）

默认情况下， Pulsar 会保存所有没确认的消息到 BookKeeper 中。持久 Topic 的消息在 Broker 重启或者 Consumer 出现问题时保存下来。

除了持久 Topic ， Pulsar 也支持非持久 Topic 。这些 Topic 的消息只存在于内存中，不会存储到磁盘。

因为 Broker 不会对消息进行持久化存储，当 Producer 将消息发送到 Broker 时， Broker 可以立即将 ack 返回给 Producer ，所以非持久 Topic 的消息传递会比持久 Topic 的消息传递更快一些。相对的，当 Broker 因为一些原因宕机、重启后，非持久 Topic 的消息都会消失，订阅者将无法收到这些消息。

3.4.3 Dead letter topic（死信主题）

死信主题允许你在用户无法成功消费某些消息时使用新消息。在这种机制中，无法使用的消息存储在单独的主题中，称为死信主题。你可以决定如何处理死信主题中的消息。

下面的例子展示了如何在 Java 客户端中使用默认的死信主题：

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES).topic(topic).subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build()).subscribe();

默认的死信主题格式：

<topicname>-<subscriptionname>-DLQ

如果你想指定死信主题的名称，请使用下面的 Java 客户端示例：

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES).topic(topic).subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).deadLetterTopic("your-topic-name").build()).subscribe();        

死信主题依赖于消息的重新投递。由于确认超时或否认确认，消息将被重新发送。如果要对消息使用否定确认，请确保在确认超时之前对其进行否定确认。
目前，在共享和 Key_Shared 订阅模式下启用了死信主题。

3.4.4 Retry letter topic（重试主题）

对于许多在线业务系统，由于业务逻辑处理中出现异常，消息会被重复消费。若要配置重新消费失败消息的延迟时间，你可以配置生产者将消息发送到业务主题和重试主题，并在消费者上启用自动重试。当在消费者上启用自动重试时，如果消息没有被消费，则消息将存储在重试主题中，因此消费者在指定的延迟时间后将自动接收来自重试主题的失败消息。

默认情况下，不启用自动重试功能。你可以将 enableRetry 设置为 true，以启用消费者的自动重试。

下面来看个如何使用从重试主题来消费消息的示例：

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES).topic(topic).subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).enableRetry(true).receiverQueueSize(100).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry").build()).subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();

3.5 Subscriptions（订阅模式）

Pulsar 支持 exclusive（独占）、failover（灾备）、 shared（共享）和 key_shared（key 共享） 四种消息订阅模式，这四种模式的示意图如下：

3.5.1 Exclusive（独占模式）

独占模式是 Pulsar 默认的消息订阅模式，在这种模式下，只能有一个 consumer 消费消息，如果有多于一个 consumer 消费此 topic 则会出错，消费示意图如下：

3.5.2 Failover（灾备模式）

灾备模式下，一个 topic 也是只有单个 consumer 消费一个订阅关系的消息，与独占模式不同之处在于，灾备模式下，每个消费者会被排序，当前面的消费者无法连接上 broker 后，消息会由下一个消费者消费，消费示意图如下：

3.5.3 Shared（共享模式）

共享模式下，消息可被多个 consumer 同时消费，无法保证消息的顺序，并且无法使用 one by one 和 cumulative 的 ack 模式，消息通过 roundrobin 的方式投递到每一个消费者，消费示意图如下：

3.5.4 Key_Shared（Key 共享模式）

Key_Shared 模式是 Shared 模式的一种，不同的是它按 key 对消息做投递，相同的 key 的消息会被投递到同一个 consumer 上，消费示意图如下：

3.6 Message retention and expiry（消息保留和过期）

默认策略：

• 立即删除所有已经被消费者确认过的的消息；
• 以 backlog 的形式，持久保存所有未被确认的消息；

两个特性：

• 消息保留让你可以保存 consumer 确认过的消息；
• 消息过期让你可以给未被确认的消息设置存活时长（TTL）；

注：所有消息保留和过期在 namespace 层面管理。

3.7 Message deduplication（消息去重）

实现消息去重的一种方式是确保消息仅生成一次，即生产者幂等。这种方式的缺点是把消息去重的工作交由应用去做。

在 Pulsar 中， Broker 支持配置开启消息去重，用户不需要为了消息去重去调整 Producer 的代码。启用消息去重后，即使一条消息被多次发送到 Topic 上，这条消息也只会被持久化到磁盘一次。

如下图，未开启消息去重时， Producer 发送消息 1 到 Topic 后， Broker 会把消息 1 持久化到 BookKeeper ，当 Producer 又发送消息 1 时， Broker 会把消息 1 再一次持久化到 BookKeeper 。开启消息去重后，当 Producer 再次发送消息 1 时， Broker 不会把消息 1 再一次持久化到磁盘。

3.7.1 去重原理

Producer 对每一个发送的消息，都会采用递增的方式生成一个唯一的 sequenceID，这个消息会放在 message 的元数据中传递给 Broker 。同时， Broker 也会维护一个 PendingMessage 队列，当 Broker 返回发送成功 ack 后， Producer 会将 PendingMessage 队列中的对应的 Sequence ID 删除，表示 Producer 任务这个消息生产成功。Broker 会记录针对每个 Producer 接收到的最大 Sequence ID 和已经处理完的最大 Sequence ID。

当 Broker 开启消息去重后， Broker 会对每个消息请求进行是否去重的判断。收到的最新的 Sequence ID 是否大于 Broker 端记录的两个维度的最大 Sequence ID，如果大于则不重复，如果小于或等于则消息重复。消息重复时， Broker 端会直接返回 ack，不会继续走后续的存储处理流程。

3.8 Delayed message delivery（消息延迟传递）

延时消息功能允许 Consumer 能够在消息发送到 Topic 后过一段时间才能消费到这条消息。在这种机制中，消息在发布到 Broker 后，会被存储在 BookKeeper 中，当到消息特定的延迟时间时，消息就会传递给 Consumer 。

下图为消息延迟传递的机制。Broker 在存储延迟消息的时候不会进行特殊的处理。当 Consumer 消费消息的时候，如果这条消息设置了延迟时间，则会把这条消息加入 DelayedDeliveryTracker 中，当到了指定的发送时间时，DelayedDeliveryTracker 才会把这条消息推送给消费者。

注：延迟消息传递仅在共享订阅模式下有效。在独占和故障转移订阅模式下，将立即分派延迟的消息。

3.8.1 示例

• Broker

  # Whether to enable the delayed delivery for messages.
  # If disabled, messages are immediately delivered and there is no tracking overhead.
  delayedDeliveryEnabled=true# Control the ticking time for the retry of delayed message delivery,
  # affecting the accuracy of the delivery time compared to the scheduled time.
  # Default is 1 second.
  delayedDeliveryTickTimeMillis=1000
  

• Producer

  // message to be delivered at the configured delay interval
  producer.newMessage().deliverAfter(3L, TimeUnit.Minute).value("Hello Pulsar!").send();
  

3.8.2 消息延迟传递原理

在 Pulsar 中，可以通过两种方式实现延迟投递。分别为 deliverAfter 和 deliverAt。

deliverAfter 可以指定具体的延迟时间戳，deliverAt 可以指定消息在多长时间后消费。两种方式本质时一样的，deliverAt 方式下，客户端会计算出具体的延迟时间戳发送给 Broker 。

DelayedDeliveryTracker 会记录所有需要延迟投递的消息的 index 。index 由 Timestamp、 Ledger ID、 Entry ID 三部分组成，其中 Ledger ID 和 Entry ID 用于定位该消息，Timestamp 除了记录需要投递的时间，还用于延迟优先级队列排序。DelayedDeliveryTracker 会根据延迟时间对消息进行排序，延迟时间最短的放在前面。当 Consumer 在消费时，如果有到期的消息需要消费，则根据 DelayedDeliveryTracker index 的 Ledger ID、 Entry ID 找到对应的消息进行消费。如下图， Producer 依次投递 m1、m2、m3、m4、m5 这五条消息，m2 没有设置延迟时间，所以会被 Consumer 直接消费。m1、m3、m4、m5 在 DelayedDeliveryTracker 会根据延迟时间进行排序，并在到达延迟时间时，依次被 Consumer 进行消费。

3.9 多租户模式

Pulsar 的云原生架构天然支持多租户，每个租户下还支持多 Namespace（命名空间），非常适合做共享大集群，方便维护。此外，Pulsar 天然支持租户之间资源的逻辑隔离，只要用户的运营管控后台和监控足够强大，便可以做到动态隔离大流量租户，防止互相干扰，还能实现大集群资源的充分利用。

• Tenant（租户）和 Namespace（命名空间）是 Pulsar 支持多租户的两个核心概念。
• 在租户级别，Pulsar 为特定的租户预留合适的存储空间、应用授权和认证机制。
• 在命名空间级别，Pulsar 有一系列的配置策略（Policy）,包括存储配额、流控、消息过期策略和命名空间之间的隔离策略。

Pulsar 的多租户性质主要体现在 Topic 的 URL 中，结构如下：

persistent://tenant/namespace/topic

租户、命名空间、topic 更直观的关系可以看下图：

3.10 统一消息模型

• Pulsar 做了队列模型与流模型的统一，在 Topic 级别只需保存一份数据，同一份数据可多次消费。以流式、队列等方式计算不同的订阅模型，大大的提升了灵活度。
• 同时 Pulsar 通过事务采用 Exactly-Once（刚好一次）的语义，在进行消息传输过程中，可以确保数据不丢不重。

3.11 Segmented Streams（分片流）

• Pulsar 将无界的数据看作是分片的流，分片分散存储在分层存储（tiered storage）、BookKeeper 集群和 Broker 节点上，而对外提供一个统一的、无界数据的视图。
• 不需要用户显式迁移数据，对用户无感知，减少存储成本并保持近似无限的存储。

3.12 Geo Replication（跨地域复制）

• Pulsar 中的跨地域复制是将 Pulsar 中持久化的消息在多个集群间备份。
• 在 Pulsar 2.4.0 中新增了复制订阅模式（Replicated-subscriptions），在某个集群失效情况下，该功能可以在其他集群恢复消费者的消费状态， 从而达到热备模式下消息服务的高可用。

在这个图中，每当 P1、P2 和 P3 生产者分别将消息发布到 Cluster-A、Cluster-B 和 Cluster-C 上的 T1 主题时，这些消息就会立即跨集群复制。一旦消息被复制，C1 和 C2 消费者就可以从他们各自的集群中消费这些消息。

没有跨地域复制，C1 和 C2 消费者就不能使用 P3 生产者发布的消息。

四、云原生架构

4.1 Pulsar 集群架构

单个 Pulsar 集群由以下三部分组成：

• 一个或者多个 broker 负责处理和负载均衡 producer 发出的消息，并将这些消息分派给 consumer；Broker 与 Pulsar 配置存储交互来处理相应的任务，并将消息存储在 BookKeeper 实例中（又称 bookies）；Broker 依赖 ZooKeeper 集群处理特定的任务，等等。
• 包含一个或多个 bookie 的 BookKeeper 集群负责消息的持久化存储。
• 一个 ZooKeeper 集群，用来处理多个 Pulsar 集群之间的协调任务。

Pulsar 分理出 Broker 与 Bookie 两层架构，Broker 为无状态服务，用于发布和消费消息，而 BookKeeper 专注于存储。Pulsar 存储是分片的，这种架构可以避免扩容时受限制，实现数据的独立扩展和快速恢复。

4.2 Brokers

Pulsar 的 broker 是一个无状态组件，主要负责运行另外的两个组件:

• 一个 HTTP 服务器（Service discovery），它暴露了 REST 系统管理接口以及在生产者和消费者之间进行 Topic 查找的 API。
• 一个调度分发器（Dispatcher），它是异步的 TCP 服务器，通过自定义二进制协议应用于所有相关的数据传输。

出于性能考虑，消息通常从 Managed Ledger 缓存中分派出去，除非积压超过缓存大小。如果积压的消息对于缓存来说太大了，则 Broker 将开始从 BookKeeper 那里读取 Entries（Entry 同样是 BookKeeper 中的概念，相当于一条记录）。

最后，为了支持全局 Topic 异地复制，Broker 会控制 Replicators 追踪本地发布的条目，并把这些条目用Java客户端重新发布到其他区域。

4.3 ZooKeeper 元数据存储

Pulsar 使用 Apache ZooKeeper 进行元数据存储、集群配置和协调。

• 配置存储 Quorum 存储了租户、命名空间和其他需要全局一致的配置项。
• 每个集群有自己独立的本地 ZooKeeper 保存集群内部配置和协调信息，例如 broker 负责哪几个主题及所有权归属元数据、broker 负载报告，BookKeeper ledger 元数据（这个是 BookKeeper 本身所依赖的）等等。

4.4 BookKeeper 持久化存储

Apache Pulsar 为应用程序提供有保证的信息传递，如果消息成功到达 broker，就认为其预期到达了目的地。

为了提供这种保证，未确认送达的消息需要持久化存储直到它们被确认送达。这种消息传递模式通常称为持久消息传递，在 Pulsar 内部，所有消息都被保存并同步 N 份，例如，2 个服务器保存四份，每个服务器上面都有镜像的 RAID 存储。

Pulsar 用 Apache BookKeeper 作为持久化存储。BookKeeper 是一个分布式的预写日志（WAL）系统，有如下几个特性特别适合 Pulsar 的应用场景：

• 使 Pulsar 能够利用独立的日志，称为 ledgers，可以随着时间的推移为 topic 创建多个 ledgers。
• 它为处理顺序消息提供了非常有效的存储。
• 保证了多系统挂掉时 ledgers 的读取一致性。
• 提供不同的 Bookies 之间均匀的 IO 分布的特性。
• 它在容量和吞吐量方面都具有水平伸缩性。能够通过增加 bookies 立即增加容量到集群中，并提升吞吐量。
• Bookies 被设计成可以承载数千的并发读写的 ledgers。 使用多个磁盘设备，一个用于日志，另一个用于一般存储，这样 Bookies 可以将读操作的影响和对于写操作的延迟分隔开。

4.4.1 brokers 与 bookies 交互

下图展示了 brokers 和 bookies 是如何交互的：

相比 Kafka、RocketMQ 等 MQ，Pulsar 基于 BookKeeper 的存储、计算分离架构，使得 Pulsar 的消息存储可以独立于 Broker 而扩展。

4.4.2 Ledgers

Ledger 是一个只追加的数据结构，并且只有一个写入器，这个写入器负责多个 BookKeeper 存储节点（就是 Bookies）的写入。 Ledger 的条目会被复制到多个 bookies。 Ledgers 本身有着非常简单的语义：

• Pulsar Broker 可以创建 ledger，添加内容到 ledger 和关闭 ledger。
• 当一个 ledger 被关闭后，除非明确的要写数据或者是因为写入器挂掉导致 ledger 关闭，这个 ledger 只会以只读模式打开。
• 最后，当 ledger 中的条目不再有用的时候，整个 legder 可以被删除（ledger 分布是跨 Bookies 的）。

4.5 Pulsar 代理

Pulsar 客户端和 Pulsar 集群交互的一种方式就是直连 Pulsar brokers 。 然而，在某些情况下，这种直连既不可行也不可取，因为客户端并不知道 broker 的地址。 例如在云环境或者 Kubernetes 以及其他类似的系统上面运行 Pulsar，直连 brokers 就基本上不可能了。

Pulsar proxy 为这个问题提供了一个解决方案，为所有的 broker 提供了一个网关，如果选择运行了Pulsar Proxy，所有的客户都会通过这个代理而不是直接与 brokers 通信。

4.6 Service discovery（服务发现）

连接到 Pulsar brokers 的客户端需要能够使用单个 URL 与整个 Pulsar 实例通信。

你可以使用自己的服务发现系统。如果你使用自己的系统，只有一个要求：当客户端端点执行 HTTP 请求，比如 http://pulsar.us-west.example.com:8080，客户端需要被重定向到一些活跃在集群所需的 broker，无论通过 DNS、HTTP 或 IP 重定向或其他手段。

五、Pulsar 相关组件

5.1 层级存储

• Infinite Stream：以流的方式永久保存原始数据
• 分区的容量不再受限制
• 充分利用云存储或现有的廉价存储（例如 HDFS）
• 数据统一表征：客户端无需关心数据究竟存储在哪里

分层存储的卸载机制就充分利用了这种面向分片式架构（segment oriented architecture）。 当需要开始卸载数据时，消息日志中的分片就依次被同步至分层存储中， 直到消息日志中所有的分片（除了当前分片之外）都已被写入分层存储后。

默认情况下写入到 BookKeeper 的数据会复制三个物理机副本。 然而，一旦分片被封存在 BookKeeper 中后，该分片就不可更改并且可以复制到归档存储中去。 长期存储可以达到节省存储费用的目的。通过使用 Reed-Solomon error correction 机制，还可减少物理备份数量。

5.2 Pulsar IO（Connector）连接器

• Pulsar IO 分为输入（Input）和输出（Output）两个模块，输入代表数据从哪里来，通过 Source 实现数据输入。输出代表数据要往哪里去，通过 Sink 实现数据输出。
• Pulsar 提出了 IO （也称为 Pulsar Connector），用于解决 Pulsar 与周边系统的集成问题，帮助用户高效完成工作。
• 目前 Pulsar IO 支持非常多的连接集成操作：例如 HDFS、Spark、Flink、Flume、ES、HBase等。

5.3 Pulsar Functions（轻量级计算框架）

• Pulsar Functions 是一个轻量级的计算框架，可以给用户提供一个部署简单、运维简单、API 简单的 FASS（Function as a service）平台。Pulsar Functions 提供基于事件的服务，支持有状态与无状态的多语言计算，是对复杂的大数据处理框架的有力补充。
• Pulsar Functions 的设计灵感来自于 Apache Storm、Apache Heron、Apache Flink 这样的流处理引擎，Pulsar Functions 将会拓展 Pulsar 和整个消息领域的未来。使用 Pulsar Functions，用户可以轻松地部署和管理 function，通过 function 从 Pulsar topic 读取数据或者生产新数据到 Pulsar topic。