1.bookkeeper 集群的架构

Apache BookKeeper 是企业级存储系统，旨在保证高持久性、一致性与低延迟。

企业级的实时存储平台需要具备的特点:

• 以极低的延迟（小于 5 毫秒）读写 entry 流
• 能够持久、一致、容错地存储数据
• 在写数据时，能够进行流式传输或追尾传输
• 有效地存储、访问历史数据与实时数据

BookKeeper 的设计完全符合以上要求，并广泛用于多种用例: 分布式系统提供高可用性或多副本,在单个集群中或多个集群间（多个数据中心）提供跨机器复制,为发布/订阅（pub-sub）消息系统提供存储服务,为流工作存储不可变对象

1.1.Bookeeper相关概念

Bookkeeper中比较核心的就两个元素: 日志(ledger/stream)和记录(entry)

1.2.记录（entry）

记录是BookKeeper中最小的I/O 单元，也被称作地址单元。单条记录中包含与该记录相关或分配给该记录的序列号（例如递增的长数）。

客户端总是从特定记录开始读取，或者追尾序列。也就是说，客户端通过监听序列来寻找下一条要添加到日志中的记录。客户端可以单次接收单条记录，也可以接收包含多条记录的数据块。序列号也可以用于随机检索记录。

1.3.日志（ledger/stream）

BookKeeper 中提供了两个表示日志存储的名词：一个是 ledger（又称日志段）；另一个是 stream（又称日志流）。

Ledger 用于记录或存储一系列数据记录（日志）。当客户端主动关闭或者当充当writer 的客户端宕机时，正在写入此 ledger 的记录会丢失，而之前存储在 ledger 中的数据不会丢失。Ledger 一旦被关闭就不可变，也就是说，不允许向已关闭的ledger 中添加数据记录（日志）。

Stream（又称日志流）是无界、无限的数据记录序列。默认情况下，stream 永远不会丢失。stream 和ledger有所不同。在追加记录时，ledger 只能运行一次，而 stream 可以运行多次。

一个 stream 由多个 ledger 组成；每个 ledger 根据基于时间或空间的滚动策略循环。在stream 被删除之前，stream 有可能存在相对较长的时间（几天、几个月，甚至几年）。Stream 的主要数据保留机制是截断，包括根据基于时间或空间的保留策略删除最早的 ledger。

Ledger 和 stream 为历史数据和实时数据提供统一的存储抽象。在写入数据时，日志流流式传输或追尾传输实时数据记录。存储在 ledger 的实时数据成为历史数据。累积在 stream 中的数据不受单机容量的限制。

1.4.命名空间

通常情况下，用户在命名空间分类、管理日志流。命名空间是租户用来创建 stream 的一种机制，也是一个部署或管理单元。用户可以配置命名空间级别的数据放置策略。

同一命名空间的所有 stream 都拥有相同的命名空间的设置，并将记录存放在根据数据放置策略配置的存储节点中。这为同时管理多个 stream 的机制提供了强有力的支持。

1.5.Bookies

Bookies 即存储服务器。一个 bookie 是一个单独的 BookKeeper 存储服务器，用于存储数据记录。BookKeeper跨 bookies 复制并存储数据 entries。出于性能考虑，单个 bookie 上存储 ledger 段，而不是整个ledger。

因此，bookie 就像是整个集成的一部分。对于任意给定 ledger L，集成指存储L 中entries 的一组bookies。将 entries 写入 ledger 时，entries 就会跨集成分段（写入 bookies 的一个分组而不是所有的bookies）。

1.6.元数据

BookKeeper 需要元数据存储服务，用来存储 ledger 与可用 bookie 的相关信息。目前，BookKeeper利用ZooKeeper 来完成这项工作（除了数据存储服务外，还包括一些协调、配置管理任务等）。

1、典型的 BookKeeper 安装包括元数据存储区（如 ZooKeeper）、bookie 集群，以及通过提供的客户端库与 bookie 交互的多个客户端。

2、为便于客户端的识别，bookie 会将自己广播到元数据存储区。

3、Bookie 会与元数据存储区交互，作为回收站收集已删除数据。

4、应用程序通过提供的客户端库与 BookKeeper 交互（使用 ledger API 或 DistributedLog Stream API）

5、应用程序 1 需要对 ledger 进行粒度控制，以便直接使用 ledger API。

6、应用程序 2 不需要较低级别 ledger 控制，因此使用更加简化的日志流 API。

1.7.Bookkeeper整体架构

Bookkeeper的元数据存储：metadata store, 目前是由zookeeper进行, 用于存储leader ID对应的元数据信息

而集群中的 bookie 用来存储这些 ledger 对应的 entry，所有的 bookie 会注册到BookKeeper 上，由客户端去发现并采取相应的操作。BookKeeper 的客户端主要是实现一些与一致性、策略性相关的逻辑。

Bookie 的实现，依靠 journal(类似于WAL预写日志) 和 ledger storage, Bookie 利用 journal 进行所有写的操作。在追加多条 entry（来自不同的 ledger）的过程中，journal都在发挥着它的持久化作用。这样做的优点是不管 ledger 来自何处，Bookie 只负责按顺序将entry写到journal文件里，不会进行随机访问。

当一个 journal 文件写满后，Bookie 会自动开启一个新的 journal 文件，继续按顺序填补 entry 。

但问题是，用户无法在 journal 里查询某条 entry。所以如果应用到读请求时，就需要「索引」功能来达到更高效的过程。

为了让各组件独立完成任务，没有在 journal 上建立索引功能，而是在 bookie 端维持了一个「write cache」，在内存里进行一个写缓存。在 journal 里运行结束后，会放置到write cache 里。

经过 write cache 过程后，Bookie对 entry 进行重新排序，按 ledger 的来源划分整理entry，以便确保在缓存变满的过程中，entry 可以按照 ledger 的顺序排队。

当缓存变满后，bookie 会把整个 write cache 冲到磁盘里。Flush 的过程又重新整理了几个目录，用来保留相关的映射关系。一个是 entry log，用来存储value。同时维护另一个ledger index，用来记录 entry id 的位置。

默认的 ledger storage 有两类：DB ledger storage 和 Sorted ledger storage。本质上，这两类ledger storage的实现途径是一样的，只是在处理索引存储时不太一样。

2.Pulsar的消息生产流程

1)客户端调用pulsar提供给客户端的API, 进行数据的生产操作, 将生产的消息传递给producer

2)在生产端内部有一个MessageWriter的类, 基于这个类实现数据分发操作, 默认方案为round-robin(轮询),同时为了提高效率, 在一定的时间内,只会选择一个partition

除了支持轮询方案外, 如果在传递消息指定key, 会采用hash取模的方式确定要发送到那个partition , 同时pulsar支持自定义分发策略

3)客户端在此连接broker, 根据要发送的partition获取对应服务的broker节点

4)broker收到消息后调用bookkeeper的客户端并发去写多个副本

5)broker端会等待bookkeeper写入完成, 当broker收到所有副本的ack之后, 会认为这条消息已经写入成功, broker会返回客户端, 告知这条消息已经被持久化完成

整个写入操作, 客户端不会跟zookeeper打交道,也不会和bookkeeper打交道, 只需要和broker即可

3.Pulsar的消息消费流程

Consumer在消费数据时候, 主要有二种情况, 一种为broker中已经缓存了消息, 一种为broker中没有缓存信息

1、消费者连接broker地址, 根据要读取的对应topic的分配, 确定要连接的最终的broker地址, 如果没有指定分片, 那么就连接每一个分片对应的broker地址

2、对应的broker首先判断消息是否已经有缓存数据, 如果存在, 直接从内存中采用推的方式发送给消费者, 将消息放置在一个receiver 队列中,消费端从队列中读取即可, 如果没有缓存,此时broker端通过bookkeeper的客户端到bookie中读取数据(内部可以读取任意副本的数据)

4.Pulsar数据读写故障处理流程

1、生产端出现失败

当出现 [发消息网络断开, broker宕机] 等情况时候, 这个时候producer有 pending 队列, 会在设置的超时时间内进行重试策略

2、Broker端出现宕机

因为broker是没有状态的, 所以它不保存任何数据, 一旦宕机后, topic的管理权会被其他broker掌管, 这个时候, 服务会被快速恢复

3、Bookkeeper出现宕机

存储节点只负责数据存储, bookkeeper本身是一个集群, 故如果只挂掉一个bookie, 并不影响, 所以broker是不会感知的,除非所有的bookie都挂掉, 没有足够的副本去写入数据.

4、消费端出现失败

一个订阅同时只有一个消费者, 但是可以拥有多个备份消费者, 一旦主消费者故障, 则备份消费者接管, 进行消费即可,同时pulsar还支持一个分区对应多个消费者, 或者一个消费端对应多个分片的情况

同时只要消息没有被消费者所消息, 在pulsar中消息就没有变成确认状态, 下次依然是可以再次消费的

5.跨机房复制

在大型的分布式系统中，都会涉及到跨多个数据中心的需求，通常会使用跨地域复制机制提供额外的冗余防止服务无法正常运作。Apache Pulsar 的跨地域多机房互备特性，是 Pulsar 企业级特性的重要组成部分，它在保证数据稳定可靠的同时，为用户提供了便捷的操作和管理。

Pulsar 自带的跨地域复制机制（Geo-Replication）可以提供一种全连接的异步复制.

在上图系统中，有三个数据中心：Cluster-A、 Cluster-B、 Cluster-C。用户创建的一个 Topic 主题 T1 设置了跨越三个数据中心做互备。在三个数据中心中，分别有三个生产者：P1、P2、P3，它们往主题 T1 中发布消息；有两个消费者：C1、C2，订阅了这个主题，接收主题中的消息。

当消息由本数据中心的生产者发布成功后，会立即复制到其他两个数据中心。消息复制完成后，消费者不仅可以收到本数据中心产生的消息，也可以收到从其他数据中心复制过来的消息。

它的工作机制是在 Broker 内部，为跨地域的数据复制启动了一组内嵌的额外生产者和消费者。当外部消息产生后，内嵌的消费者会读取消息；读取完成后，调用内嵌的生产者将消息立即发送到远端的数据中心。

跨地域复制需要设置“租户”在数据中心之间的访问权限。

在配置了跨地域复制后，每个发送进来的消息，首先被保存在本地集群中；然后异步地推送到远端的集群。如果本地集群和远端集群之间没有网络问题，消息会被立即推送给远端集群。这样端到端的发送延迟主要由集群之间网络的决定。

在图示中，无论生产者（Producer）P1、P2 和 P3 在什么时候分别将消息发布给 Cluster A、 Cluster B 和 Cluster C 中的主题T1，这些消息均会立刻复制到整个集群。一旦完成复制，消费者（Consumer）C1 和 C2 即可从自己所在的集群消耗这些消息，并且保持消息在每个Producer 内部的发送顺序。

因为消息已经从其他远端集群发送到本地集群的 Topic，所以每个集群内部都会保留这个 Topic 中产生的所有消息。对于每个Consumer 来说，Consumer 的订阅（subscription，维护 Consumer 对 Topic 的消费和 ack 的位置）是针对本地集群的 Topic，相当于 Consumer 消费本地集群的消息。

5.1.配置操作

第一步：创建一个租户，并给予三个数据中心的权限

$ bin/pulsar-admin tenants create my-tenant \
--allowed-clusters cluster1, cluster2, cluster3

第二步：创建namespace

$ bin/pulsar-admin namespaces create my-tenant/my-namespace

第三步：设置namespace中topic在哪些数据中心之间进行互备

$ bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace \
--clusters cluster1, cluster2, cluster3

说明: 在未来, 如果新增了数据中心,或者关闭数据中心, 可以随时进行配置调整操作, 而且pulsar表示这样的操作并不会对流量有任何影响

以下模拟了一个新增数据中心的操作:

bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace \
--clusters cluster1, cluster2, cluster3

转载请注明：西门飞冰的博客 » Pulsar 架构原理