Pulsar

一、简介

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

二、架构介绍

Pulsar由Producer、Consumer、多个Broker、一个BookKeeper集群、一个Zookeeper集群构成

Producer：数据生成者，即发送消息的一方。生产者负责创建消息，将其投递到Pulsar中。

Consumer：数据消费者，即接收消息的一方。消费者连接到 Pulsar 并接收消息，进行相应的业务处理。

Broker：无状态的服务层，负责接收消息、传递消息、集群负载均衡等操作，Broker不会持久化保存元数据。

BookKeeper：有状态的持久层，包含多个Bookie，负责持久化地存储消息。

ZooKeeper：存储Pulsar、BookKeeper的元数据，集群配置等信息，负责集群间的协调(例如：Topic与Broker的关系)、服务发现等。

Broker扩展

在Pulsar中Broker是无状态的，当需要支持更多的消费者或生产者时，可以简单地添加更多的Broker节点来满足业务需求。Pulsar支持自动的分区负载均衡，在Broker节点的资源使用率达到阈值时，会将负载迁移到负载较低的Broker节点，这个过程中分区也将在多个Broker节点中做平衡迁移，一些分区的所有权会转移到新的Broker节点。

Bookie扩展

存储层的扩容，通过增加Bookie节点来实现。在BooKie扩容的阶段，由于分片机制，整个过程不会涉及到不必要的数据搬迁，即不需要将旧数据从现有存储节点重新复制到新存储节点。

Topic

1. 分区Topic（Topic-Partition）

Pulsar的Topic可以分为非分区Topic和分区Topic。普通的Topic仅仅被保存在单个Broker中，这限制了Topic的最大吞吐量。分区Topic是一种特殊类型的主题，支持被多个Broker处理，从而实现更高的吞吐量。

2. 持久Topic、非持久Topic

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

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

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

3. 重试Topic

由于业务逻辑处理出现异常，消息一般需要被重新消费。Pulsar支持生产者同时将消息发送到普通的Topic和重试Topic，并指定允许延时和最大重试次数。当配置了允许消费者自动重试时，如果消息没有被消费成功，会被保存到重试Topic中，并在指定延时时间后，重新被消费。

4. 死信Topic

当Consumer消费消息出错时，可以通过配置重试Topic对消息进行重试，但是，如果当消息超过了最大的重试次数仍处理失败时，该怎么办呢？Pulsar提供了死信Topic，通过配置deadLetterTopic，当消息达到最大重试次数的时候，Pulsar会将消息推送到死信Topic中进行保存。

订阅（subscription）

订阅类型（Subscription type）

Pulsar支持独占（Exclusive）、灾备（Failover）、共享（Shared）、Key_Shared这四种订阅类型。

1. 独占（Exclusive）SinglePartition

Exclusive下，只允许Subscription存在一个消费者，如果多个消费者使用同一个订阅名称去订阅同一个Topic，则会报错。如下图，只有Consumer A-0可以消费数据。

2. 灾备（Failover）

Failover下，一个Subscription中可以有多个消费者，但只有Master Consumer可以消费数据。当Master Consumer断开连接时，消息会由下一个被选中的Consumer进行消费。

分区Topic：Broker会按照消费者的优先级和消费名的顺序对消费者进行排序，将Topic均匀地分配给优先级最高的消费者。

非分区Topic：Broker会根据消费者订阅的非分区Topic的时间顺序选择消费者。

3. 共享（Shared）

shared中，多个消费者可以绑定到同一个Subscription上。消息通过 round robin即轮询机制分发给不同的消费者，并且每个消息仅会被分发给一个消费者。当消费者断开连接，所有被发送给消费者但没有被确认的消息将被重新处理，分发给其它存活的消费者。

4. Key_Shared

Key_Shared中，多个Consumer可以绑定到同一个Subscription。消息在传递给Consumer时，具有相同键的消息只会传递给同一个Consumer。

订阅模式（Subscription modes）

订阅模式有持久化和非持久化两种。订阅模式取决于游标(cursor)的类型。

创建订阅时，将创建一个相关的游标来记录最后使用的位置。当订阅的consumer重新启动时，它可以从它所消费的最后一条消息继续消费。

Durable（持久订阅）：游标是持久性的，会保留消息并保持游标记录的位置。当Broker重新启动时，可以从BookKeeper中恢复游标，消息可以从游标上次记录的位置继续消费。默认情况下，都是持久化订阅。

NonDurable（非持久订阅）：游标不是持久性的，当Broker宕机时，游标会丢失并无法恢复，所以消息无法继续从上次消费的位置开始继续消费。

多主题订阅

当Consumer订阅Topic时，默认指定订阅一个主题。从Pulsar的1.23.0-incubating的版本开始，Pulsar消费者可以同时订阅多个Topic。可以通过两种方式进行订阅：

正则表达式，例如：

persistent://public/default/finance-.*

明确指定Topic列表。

Pulsar生产者（Producer）

1. 访问模式

消息生成者有多种模式访问Topic ，可以使用以下几种方式将消息发送到Topic。

• Shared：默认情况下，多个生成者可以将消息发送到同一个Topic。

• Exclusive：在这种模式下，只有一个生产者可以将消息发送到Topic ，当其他生产者尝试发送消息到这个Topic时，会发生错误。只有独占Topic的生产者发生宕机时（Network Partition）该生产者会被驱逐，新的生产者才能产生并向Topic发送消息。

• WaitForExclusive：在这种模式下，只有一个生产者可以将消息发送到Topic。当已有生成者和Topic建立连接时，其他生产者的创建会被挂起而不会产生错误。如果想要采用领导者选举机制来选择消费者的话，可以采用这种模式。

2.路由模式

当将消息发送到分区Topic时，需要指定消息的路由模式，这决定了消息将会被发送到哪个分区Topic。Pulsar有以下三种消息路由模式，RoundRobinPartition为默认路由模式。

• RoundRobinPartition：如果消息没有指定key，为了达到最大吞吐量，生产者会以round-robin （轮询）方式将消息发布到所有分区。请注意round-robin并不是作用于每条单独的消息，而是作用于延迟处理的批次边界，以确保批处理有效。如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。这是默认的模式。

• SinglePartition：如果消息没有指定key，生产者将会随机选择一个分区，并发布所有消息到这个分区。如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。

• CustomPartition：自定义模式，用户可以创建自定义路由模式，通过实现MessageRouter接口实现自定义路由规则。

批量处理

Pulsar支持对消息进行批量处理。批量处理启用后，Producer会在一次请求中累积并发送一批消息。批量处理时的消息数量取决于最大消息数（单次批量处理请求可以发送的最大消息数）和最大发布延迟（单个请求的最大发布延迟时间）决定。

索引确认机制

启用批量索引确认机制，Consumer将筛选出已被确认的批量索引，并将批量索引确认请求发送给Broker。Broker维护批量索引的确认状态并跟踪每批索引的确认状态，以避免向Consumer发送已确认的消息。当该批信息的所有索引都被确认后，该批信息将被删除。

key-based batching

key_shared模式下，Broker会根据消息的key来分发消息，但默认的批量处理模式，无法保证将所有的相同的key都打包到同一批中，而且Consumer在接收到批数据时，会默认把第一个消息的key当作这批消息的key，这会导致消息的错乱。因此key_shared模式下，不支持默认的批量处理。

key-based batching能够确保Producer在打包消息时，将相同key的消息打包到同一批中，从而consumer在消费的时候，也能够消费到指定key的批数据。

没有指定key的消息在打包成批后，这一批数据也是没有key的，Broker在分发这批消息时，会使用NON_KEY作为这批消息的key。

Pulsar消费者（Consumer）

Pulsar提供两种确认模式：

• 累积确认：消费者只需要确认最后一条收到的消息，在此之前的消息，都不会被再次发送给消费者。

• 单条确认：消费者需要确认每条消息并发送ack给Broker。

AcknowledgmentsGroupingTracker

消息的单条确认和累积确认并不是直接发送确认请求给Broker，而是把请求转交给AcknowledgmentsGroupingTracker处理。

为了保证消息确认的性能，并避免Broker接收到非常高并发的ack请求，Tracker默认支持批量确认，即使是单条消息的确认，也会先进入队列，然后再一批发往Broker。在创建consumer的时候，可以设置acknowledgementGroupTimeMicros，默认情况下，每100ms或者堆积超过1000时，AcknowledgmentsGroupingTracker会发送一批确认请求。如果设置为0，则每次确认消息后，Consumer都会立即发送确认请求。

Pulsar服务端

Broker是Pulsar的一个无状态组件，主要负责运行以下两个组件：

http服务：提供为生产者和消费者管理任务和Topic查找的REST API。Producer通过连接到Broker来发送消息，Consumer通过连接到Broker来接收消息。

调度器：提供异步http服务，用于二进制数据的传输。

Pulsar Broker会默认删除已经被所有Consumer确认的消息，并以backlog的方式持久化存储所有未被确认的内消息。Pulsar的message retention(消息留存) 和message expiry (消息过期)这两个特性可以调整Broker的默认设置。

Message retention: 保留Consumer已确认的消息。

通过留存规则的设定，可以保证已经被确认且符合留存规则的消息持久地保存在Pulsar中，而没有被留存规则覆盖、已经被确认的消息会被删除。

Message expire（消息过期）：设置未确认消息的存活时长（TTL）。

通过设置消息的TTL，有些即使还没有被确认，但已经超过TTL的消息，也会被删除

消息去重

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

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

去重原理

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，不会继续走后续的存储处理流程。

Bundle

Topic分区与Broker的关联是通过Bundle机制进行管理的。

每个namespace存在一个Bundle列表，在namesapce创建时可以指定Bundle的数量。Bundle其实是一个分片机制，每个Bundle拥有 namespace 整个hash范围的一部分。每个Topic (分区) 通过hash运算落到相应的Bundle区间，进而找到当前区间关联的Broker。每个Bundle绑定唯一的一个Broker，但一个Broker可以有多个Bundle。

Pulsar存储层（Bookkeeper）

分片存储

概念：

• Bookie：BookKeeper的一部分，处理需要持久化的数据。

• Ledger：BookKeeper的存储逻辑单元，可用于追加写数据。

• Entry：写入BookKeeper的数据实体。当批量生产时，Entry为多条消息，当非批量生产时，Entry为单条数据