[大数据]Apache Pulsar基本理论

前言

参考链接：
Apache Pulsar 官网
知乎 pulsar的架构与核心概念
CSDN 什么是Pulsar
CSDN Apache Pulsar之与Apache Kafka的异同
掘金 Pulsar从入门到实现

什么是Pulsar？

Pulsar是由雅虎创建的开源的、 是一个用于服务器到服务器的消息系统，具有多租户、高性能等优势。现在是Apache基金会的一个孵化项目。

1 Pulsar特性

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

Pulsar的关键特性：

2 Pulsar消息队列

2.1 生产者

生产者是一个附加到topic并将消息发布到 Pulsar broker的进程。 Pulsar broker 处理消息。

2.1.1 发送模式

Producer 可以以同步(sync) 或 异步(async) 的方式发布消息到 broker

2.1.2 访问模式

对于消息生产者来说在主题上你可以有不同类型的访问模式

2.1.3 压缩

Pulsar 生产者目前支持以下类型的压缩：

• LZ4
• ZLIB
• ZSTD
• SNAPPY

2.1.4 批量处理

当批量处理启用时，producer 会在单个请求中积累并发送一批消息。 批量处理的量大小由最大消息数和最大发布延迟定义。 因此，积压数量是分批处理的总数，而不是信息总数。
在 Pulsar中，批次被跟踪并存储为单个单元，而不是单个消息。Consumer将批量处理的消息拆分成单个消息。但即使启用了批量处理，也始终将计划中的消息(通过 deliverAt 或者 deliverAfter 进行配置) 作为单个消息发送。
一般来说，当consumer确认了一个批的所有消息，该批才会被认定为确认。这意味着当发生不可预料的失败、取消确认（negative acknowledgements）或确认超时，都可能导致批中的所有消息都被重新发送，即使其中一些消息已经被确认了。
为了避免将确认的消息批量重新发送给消费者，Pulsar从Pulsar 2.6.0开始引入了批量索引确认。当启用批量索引确认时，消费者过滤掉已经确认的批量索引，并向broker发送批量索引确认请求。Broker 维护批量索引的确认状态并跟踪每批索引的确认状态，以避免向consumer发送已确认的消息。当某一批消息的所有索引都被确认时，该批消息将被删除。
批量索引确认默认是关闭的(acknowledgmentAtBatchIndexLevelEnabled=false)

2.1.5 分块（chunking）

1. 分块有以下特性：

• 不能同时启用批处理和分块
• 分块仅仅支持持久化topic
• 分块仅仅支持exclusive和failover订阅模式

当启用分块时(chunkingEnabled=true) ，如果消息大小大于允许的最大发布有效载荷大小，则 producer 将原始消息分割成分块的消息，并将它们与块状的元数据一起单独和按顺序发布到 broker。 在 broker 中，分块的消息将和普通的消息以相同的方式存储在 Managed Ledger 上。 唯一的区别是，consumer 需要缓冲分块消息，并在收集完所有分块消息后将其合并成真正的消息。 Managed Ledger上的分块消息可以和普通消息交织在一起。 如果 producer 未能发布消息的所有分块，则当 consumer 未能在过期时间(expire time) 内接收所有分块时，consumer可以过期未完成的分块。默认情况下，过期时间设置为1小时。
Consumer 会缓存收到的块状消息，直到收到消息的所有分块为止。 然后 consumer 将分块的消息拼接在一起，并将它们放入接收器队列中。 客户端从接收器队列中消费消息。 一旦 consumer 使用整个大消息并确认，consumer 就会在内部发送与该大消息关联的所有分块消息的确认。当达到阈值(maxPendingChunkedMessage)时，consumer通过静默确认未分块的消息或通过将其标记为未确认，要求 broker 稍后重新发送这些消息。
非Shared模式下，broker不需要任何更改来支持分块。broker使用(chunkedMessageRate) 来记录topic上的分块消息速率。

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

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

3. 多个生产者和一个生产者处理块消息

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

2.2 消费者

在Consumer端有一个队列，用于接收从broker推送来的消息。通过receiverQueueSize参数配置队列的长度 (队列的默认长度是1000) 每当 consumer.receive() 被调用一次，就从缓冲区（buffer）获取一条消息。

2.2.1 接收模式

可以通过同步(sync) 或者异步(async)的方式从brokers接受消息。

2.2.2 确认

当消费者成功的消费了一条消息，这个消费者会发送一个确认信息给broker。 这个消息时是永久保存的，只有在收到订阅者消费成功的消息确认后才会被删除。 如果希望消息被 Consumer 确认后仍然保留下来，可配置消息保留策略实现。
对于批消息，如果批量索引确认是打开的，broker会维护批索引确认状态并跟踪每个批索引的确认状态以避免将确认消息分发给消费者。当某一批消息的所有索引都被确认时，该批消息将被删除。
可以通过以下两种方式确认消息：

• 消息被单独确认。 通过单独确认，消费者需要确认每条消息并向broker发送确认请求
• 累积确认模式 累积确认时，消费者只需要确认最后一条他收到的消息。 所有之前（包含此条）的消息，都不会被再次发送给那个消费者。

2.2.3 取消确认

当消费者在某个时间没有成功的消费某条消息，消费者想重新消费到这条消息，这个消费者可以发送一条取消确认消息到broker，broker会将这条消息重新发给消费者。
消息取消确认也有单条取消模式和累积取消模式 ，这依赖于消费者使用的订阅模式。
在exclusive模式和failover模式中，消费者仅仅只能对收到的最后一条消息进行取消确认。
在 shared和Key_Shared模式下，可以单独取消确认消息。

2.2.4 确认超时

如果消息没有被成功消费，你想去让 broker 自动重新交付这个消息， 你可以采用未确认消息自动重新交付机制。 客户端会跟踪 超时 时间范围内所有未确认的消息。 并且在指定超时时间后会发送一个 重发未确认的消息 请求到 broker。

2.2.5 死信主题

死信主题使您可以在消费者无法成功消费某些消息时消费新消息。 在这种机制中，消费失败的消息存储在一个单独的主题中，称为死信主题。 您可以决定如何处理死信主题中的消息。
以下示例显示如何使用默认死信主题在 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();

死信主题取决于消息重新传递。 由于确认超时或取消确认，消息被重新传送。 如果打算对消息使用取消确认，请确保在确认超时之前对其进行取消确认。

2.2.6 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();

2.3 topic

2.3.1 分区topic

单个topic的消息一般是由单个broker处理，为了提高topic的消息处理能力，pulsar提供了partitioned topic的支持，与kafka和rocketmq一样，每个partition由不同的broker处理，在消费时，单个partition可选择exclusive, failover和shared模式
Partitioned topic实际上是由n（partition的数量）个内部的topic组成的，每个内部的topic由一个broker处理，每个broker可处理多个topic，当消息发送到broker前，在producer端会通过routing mode将消息路由到某一个partition上，消息的生产与消费示意图如下：

2.3.2 消息的保留和过期

默认情况上，当broker会立刻删除所有收到了ack的消息，没有被ack的消息会持久化存储，但是我们可以修改pulsar的行为，pulsar允许我们存储已经收到ack了的消息，也可以给未收到ack的消息设置过期时间（TTL）

2.3.3 消息去重

Pulsar支持在broker端对消息做去重，当打开消息去重后，重发的消息（重试等产生的）不会被重新存储，这个特性使得pulsar对流式计算引擎（例如flink）更加友好，流式计算引擎更容易实现exactly-once语义的计算任务，消息去重的存储示意图如下：

2.3.4 消息延迟传递

延时消息功能允许你能够过一段时间才能消费到这条消息，而不是消息发布后，就马上可以消费到。
延迟消息传递仅适用于Shared模式。 在 Exclusive 和 Failover 订阅模式下，延迟消息会立即发送。
如下图所示，说明了延时消息的实现机制：

默认情况下启用延迟消息传递。 您可以在代理配置文件中更改它

$ 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

下面是 Java 当中生产延时消息一个例子

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

2.4 命名空间

命名空间是租户内部逻辑上的命名术语。 可以通过admin API在租户下创建多个命名空间。 例如，包含多个应用程序的租户可以为每个应用程序创建单独的命名空间。 Namespace使得程序可以以层级的方式创建和管理topic Topicmy-tenant/app1 ，它的namespace是app1这个应用，对应的租户是 my-tenant。 你可以在namespace下创建任意数量的topic。

2.5 订阅

Pulsar支持exclusive、shared和failover三种消息订阅模式，这三种模式的示意图如下：

1. Exclusive模式（独占模式）

pulsar默认的消息订阅模式，在这种模式下，中能有一个consumer消息消息，一个订阅关系中只能有一台机器消费每个topic，如果有多于一个consumer消费此topic则会出错，消费示意图如下：

2. Failover模式

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

3. Shared模式（共享模式）

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

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

3 Pulsar的架构

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

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

下图为一个 Pulsar 集群：

集群通过ZooKeeper来进行协处理，比如异地灾备、异地复制。

3.1 Broker

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

• 一个 HTTP 服务器，它为生产者和消费者公开管理任务和topic查找的 REST API。 生产者连接到broker发布消息，消费者连接到broler来消费消息。
• 一个调度分发器, 它是异步的TCP服务器，通过自定义 二进制协议应用于所有相关的数据传输。
  出于性能考虑，消息通常从managed ledger缓存中分派，除非积压超过缓存大小。如果积压的消息对于缓存来说太大了, 则Broker将开始从BookKeeper那里读取Entries（Entry同样是BookKeeper中的概念，相当于一条记录）。

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

3.2 Cluster

一个 Pulsar 实例由一个或多个 Pulsar 集群组成。 反过来，集群包括：

• 一个或者多个Pulsar brokers
• 一个ZooKeeper协调器，用于集群级别的配置和协调
• 一组BookKeeper的Bookies用于消息的 持久化存储

集群间可以通过异地复制进行消息同步。

3.3 元数据存储

Pulsar 元数据存储维护一个 Pulsar 集群的所有元数据，例如topic元数据、schema、broker加载数据等。Pulsar 使用 Apache ZooKeeper 进行元数据存储、集群配置和协调。Pulsar 元数据存储可以部署在单独的 ZooKeeper 集群上，也可以部署在现有的 ZooKeeper 集群上。您可以将一个 ZooKeeper 集群同时用于 Pulsar 元数据存储和 BookKeeper 元数据存储。如果要部署连接到现有 BookKeeper 集群的 Pulsar broker，则需要分别为 Pulsar 元数据存储和 BookKeeper 元数据存储部署单独的 ZooKeeper 集群。
在一个Pulsar实例中：

• 配置存储仲裁存储租户、命名空间和其他需要全局一致的实体的配置
• 每个集群都有自己的本地 ZooKeeper 集成，用于存储特定于集群的配置和协调，例如哪些代理负责哪些topic以及所有权元数据、broker加载报告、BookKeeper ledger元数据等。

存储配置
存储配置维护一个 Pulsar 实例的所有配置，例如集群、租户、命名空间、分区topic相关配置等。一个Pulsar实例可以有一个本地集群、多个本地集群或多个跨区域集群。因此，配置存储可以在Pulsar实例下的多个集群之间共享配置。配置存储可以部署在单独的 ZooKeeper 集群上，也可以部署在现有的 ZooKeeper 集群上。

3.4 持久化存储

Pulsar 为应用程序提供有保证的消息传递。 如果消息成功到达 Pulsar broker，它将被传递到其预期目标。
为了提供这种保证，未确认送达的消息需要持久化存储直到它们被确认送达。这种消息传递模式通常称为持久消息传递。所有消息都被保存并同步N份，例如，2个服务器保存四份，每个服务器上面都有镜像的RAID存储。

3.4.1 Apache Bookkeeper

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

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

除了消息数据，cursors也持久存储在 BookKeeper 中。Cursors是消费端订阅消费的位置。 BookKeeper让Pulsar可以用一种可扩展的方式存储消费位置。
目前，Pulsar 支持持久化消息存储。 这说明了所有topic名称中的持久性。下面是一个示例：

persistent://tenant/namespace/topic

Pulsar也支持临时消息（ (non-persistent) ）存储

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

3.4.2 Ledgers

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

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

3.4.3 Ledger读一致性

BookKeeper的主要优势在于他能在有系统故障时保证读的一致性。 由于Ledger只能被一个进程写入（之前提的写入器进程），这样这个进程在写入时不会有冲突，从而写入会非常高效。 在一次故障之后，ledger会启动一个恢复进程来确定ledger的最终状态并确认最后提交到日志的是哪一个条目。 在这之后，能保证所有的ledger读进程读取到相同的内容。

3.4.4 ledgers管理（managed ledger）

鉴于 Bookkeeper Ledger提供单一日志抽象，在ledger之上开发了一个库，称为managed ledger，代表单个主题的存储层。managed ledger即消息流的抽象，有一个写入器进程不断在流结尾添加消息，并且有多个cursors 消费这个流，每个cursor有自己的消费位置。
在内部，单个managed ledger使用多个 BookKeeper ledgers来存储数据。 有多个ledgers的原因有两个：

• 在故障之后，原有的某个ledger不能再写了，需要创建一个新的。
• 当所有cursors都消耗了它包含的消息时，可以删除ledger。 这允许ledger的定期滚动。

3.5 日志存储

在 BookKeeper 中，日志文件包含 BookKeeper 事务日志。在更新到 ledger之前，bookie需要确保描述这个更新的事务被写到持久（非易失）存储上面。 在bookie启动和旧的日志文件大小达到上限（由 journalMaxSizeMB 参数配置）的时候，新的日志文件会被创建。

3.6 Pulsar协议

Pulsar客户端和Pulsar集群交互的一种方式就是直连Pulsar brokers 。 然而，在某些情况下，这种直连既不可行也不可取，因为客户端并不知道broker的地址。 例如在云环境或者 Kubernetes 以及其他类似的系统上面运行Pulsar，直连brokers就基本上不可能了。
Pulsar 协议通过充当集群中所有broker的单一网关来解决此问题。如果你选择运行Pulsar Proxy（这是可选的），所有的客户端连接将会通过这个代理而不是直接与brokers通信。
架构上来看，Pulsar Proxy从ZooKeeper上面读取他所需要的所有信息。 当启动代理时，你只需要提供用于集群独有和实例范围的配置存储的ZooKeeper连接串。 下面是一个示例：

$ bin/pulsar proxy \
  --zookeeper-servers zk-0,zk-1,zk-2 \
  --configuration-store-servers zk-0,zk-1,zk-2

关于Pulsar proxy有一些比较重要的注意点：

• 连接客户端不需要提供任何特定的配置来使用 Pulsar 协议。除了更新用于服务URL的IP之外，你不需要为现有的应用更新客户端配置（例如你在Pulsar proxy上层架设运行了负载均衡器）。
• Pulsar proxy支持TLS 加密 和 认证。

4 Pulsar与kafka的区别

4.1 性能对比

Pulsar 表现最出色的就是性能，Pulsar 的速度比 Kafka 快得多，与 Kafka 相比，Pulsar 的速度提升了 2.5 倍，延迟降低了 40％。


注：对比是针对 1 个分区的 1 个主题，其中包含 100 字节消息，Pulsar 每秒可发送 220,000+ 条消息。

4.2 对比总结

TTL是订阅的消费时间限制。如果在配置的TTL时间段内没有任何消费者使用消息，则消息将自动标记为已确认。