小县城做服务网站,wordpress文章页的三大标签,网站建设前期情况说明,动漫制作需要学什么导语
Apache Pulsar 是一个多租户、高性能的服务间消息传输解决方案#xff0c;支持多租户、低延时、读写分离、跨地域复制#xff08;GEO replication#xff09;、快速扩容、灵活容错等特性。Pulsar 存储层依托于 BookKeeper 组件#xff0c;所以本文简单探讨一下 BookK…导语
Apache Pulsar 是一个多租户、高性能的服务间消息传输解决方案支持多租户、低延时、读写分离、跨地域复制GEO replication、快速扩容、灵活容错等特性。Pulsar 存储层依托于 BookKeeper 组件所以本文简单探讨一下 BookKeeper(下文简称 BK) 的一致性协议是如何实现的。
背景
Pulsar相对于 Kafka 根本的区别在于数据一致性协议这也是为什么 Pulsar 可以做到两副本就能保障高可用、高可靠在磁盘使用方面更均衡也不会存在单分区容量上限同时在扩缩容、故障屏蔽等日常运维方面更加灵活和方便。
一致性协议简介
我们常见的一致性协议比如 Raft、Kafka、ZAB 等都是服务端集成协议(协议控制和数据存储绑定)简单来说一致性协议由服务端存储节点来执行。数据流向通常是客户端写数据到 Leader 节点其他节点再通过推或拉的方式从 Leader 获取数据。 而 BK 的一致性协议控制和存储是分开的协议控制是在客户端执行可以称之为外部一致性协议或者客户端一致性协议。数据流向为客户端向多台存储节点同时写入数据存储节点之间基本不通信。 由于一致性协议主要是在客户端执行本文聚焦于 BK 客户端的实现。所以外部一致性协议需要解决的问题就简化成给定 N 个 KV 存储(Bookie 高性能、低可靠)和一个分布式协调系统下文以 ZK 为例, 低性能高可靠如何实现高可用、高可靠、低时延的读写服务。这里引入 ZK 集群的原因是 Bookie 节点会动态增减需要有注册中心能让客户端拉取存活 Bookie 的 IPPort同时客户端不具备存储能力协议控制必要的元信息信息也需要存储到外部。在详细了解 BK 一致性协议之前 我们先简单介绍一下BK 提供的基础能力。
BookKeeper 基本能力
协议作者设计 BK 初衷是提供分布式日志段Ledger而不是无界日志如 Raft、Kafka。所以 BK 提供是主要能力就是创建删除 Ledger 以及 Ledger 的读写操作。代码片段如下:
public interface BookKeeper extends AutoCloseable {CreateBuilder newCreateLedgerOp(); // 创建新 LedgerOpenBuilder newOpenLedgerOp(); // 打开已存在LedgerDeleteBuilder newDeleteLedgerOp(); // 删除 Ledger}
public interface ReadHandle extends Handle {CompletableFutureLedgerEntries readAsync(long firstEntry, long lastEntry); // 从 ledger 里读数据}public interface WriteHandle extends ReadHandle, ForceableHandle {CompletableFutureLong appendAsync(ByteBuf data); // 往ledger 写数据
}而 MQ 场景需要提供无界数据流所以 Pulsar 为每一个分区维护一组 Ledger每个 Ledger 只会写一定量数据条数 | 大小 | 时间限制写满就会创建新的 Ledger 来继续写入只有最新的 Ledger 可以写入历史的 Ledger 只能读取。再根据用户配置的数据保存策略逐步删除历史 Ledger。Pulsar 内分区 Legder 组成如下: Pulsar 内分区的元数据如下:
{lastLedgerCreatedTimestamp: 2024-06-06T17:03:54.66608:00, waitingCursorsCount: 0,pendingAddEntriesCount: 2,lastConfirmedEntry: 11946613:185308,state: LedgerOpened,ledgers: [{ledgerId: 11945498,entries: 194480,size: 420564873,offloaded: false,underReplicated: false
}, {ledgerId: 11946057,entries: 194817,size: 420583702,offloaded: false,underReplicated: false
}, {ledgerId: 11946613,entries: 0, // 未关闭的ledger还没有条数size: 0, // 未关闭的ledger还没有大小 offloaded: false,underReplicated: false
}]
}创建 Ledger
BK 以 Ledger 维度对外提供服务第一步就是创建 Ledger。一致性协议高可用是由数据的多副本存储来实现的所以创建 Ledger 时需要明确指定该 Ledger 的数据存储在几台 Bookie(E)每条数据写几份Qw需要等到多少台 Bookie 确认收到数据后才认为数据写入成功Qa。代码片段如下:
public interface CreateBuilder extends OpBuilderWriteHandle {// E: 接收该 Ledger 数据的节点数 默认值3CreateBuilder withEnsembleSize(int ensembleSize);/*** Qw: 每条数据需要写几个节点, 默认值2* 如果 E Qw, 数据会条带化均匀写入到 Bookie 节点按照默认配置* 写入顺序为: 消息00,1; 消息11,2; 消息32,0*/CreateBuilder withWriteQuorumSize(int writeQuorumSize);// Qa: 需要等待几台节点确认写入成功默认值2CreateBuilder withAckQuorumSize(int ackQuorumSize);ps: 配置条带化写入E Qw可以让多个节点分摊该 Ledger 的读写压力但是由于数据会更分散且不连续会影响服务端读写优化顺序写、预读、批读等 所以生产环境建议配置 EQw。而 Qa Qw 可以让客户端在避免等待慢节点返回从而降低写入时延但是优化效果有限同时增加了存储成本可以配合客户端熔断以及服务端指标来剔除慢节点。简单来说建议生产环境配置 EQwQa。
在创建 Ledger 时客户端会从 ZK 中拉取全部 Bookie根据 EnsemblePlacementPolicy 放置策略挑选节点默认策略会随机挑选出 E 个数的节点作为初始存储节点以保证存储节点使用均衡。同时 BK 提供丰富的放置策略满足用户实现跨机房、地区容灾的需求。 现在我们已经有了 Ledger 基本元数据包含 Ledger 配置信息以及初始的 Bookie 列表由于 Ledger 需要全局唯一所以还需要使用 ZK 获取一个全局唯一的ID最后把全部元数据写入到 ZK 集群即可(创建过程无需与 Bookie 节点交互)。代码如下: BK 中 Ledger 元数据信息如下:
LedgerMetadata { formatVersion 3, ensembleSize 3, writeQuorumSize 3, ackQuorumSize 3, state CLOSED, length 461178392, lastEntryId 203045, digestType CRC32C, password base64: , ensembles {0 [127.0 .0 .1: 3181, 127.0 .0 .2: 3181, 127.0 .0 .3: 3181] }, customMetadata { component base64: bWFYWdlZC1sZWRnZXI , pulsar / managed - ledger base64: cHVibGjL2RlZmF1bHQvcGVyc2lzdGVudC9kd2RfMjEwODQtcGFydGl0aW9uLTA , application base64: cHVsc2Fy }
}数据写入
Ledger 已经创建好了现在可以往里面写数据了。由于 Bookie 是个 KV 存储数据需要全局唯一的 Key, Ledger Id 已经是全局唯一了并且只有创建这个 Ledger 的客户端才能往里面写数据所以数据的 Key 为 LedgerId EntryId客户端单调自增即可。 每次写入都从 E 个Bookie 中顺序挑选出 Qw 个节点writeSet并行发送数据。 可以看到客户端并行写入 Bookie 节点Bookie 间也不需要做数据同步所以 Pulsar 即使加了一层 Broker 层但是写入时延还是能做到和 Kafka 基本一致2RTT。
正常返回
发送数据后如果收到 Qa 个数成功响应 就可以认为这条数据写入成功了。同时这条数据也可以被读取。 LAC(LastAddConfirmed)
由于 EntryId 单调递增数据按顺序写入当一个位置的数据成功写入到 Qa 个存储节点后就意味着该条数据以及之前的所有数据都可以被消费走这个位置就是可读位置不考虑事务由于已写入多副本也意味着这些数据可以在部分存储节点故障下存活从而保证读一致性。不同协议中有不同的叫法。在 BK 中这个位置称为 LACLastAddConfirmedKafka 中称为 HWHigh Watermark**在 Raft 中称为 Commitindex。
下面代码可以看到成功写入一条数据后会立即更新可读位置。 LAC 是一致性协议至关重要的信息正常情况下LAC 只需要维护在内存中写入成功后更新。读取时使用 LAC 来限制读取位置即可。如果当前 Ledger 写满了正常关闭需要把 LAC 也保存在 ZK 的元数据中。这样重新打开 Ledger 用于读取时才能加载 LAC 回内存中用来正确限制读取位置同时 LAC 值也能标识 Ledger 中存储的数据条数。
而异常情况下故障会直接导致内存中的 LAC 丢失这是不可接受的。最简单粗暴的方式是每次更新 LAC 都直接同步到 ZK 中但显然 ZK 性能无法满足高频率写入所以 BK 的做法是在每条数据都携带 LAC 信息一起写入到 Bookie 中。 如下图由于最新的 LAC 需要等下一轮的消息写入才会一起存储到 Bookie所以 Bookie 中存储的 LAC 与实际的 LAC 相比是存在一定滞后。 上图可看到初始时客户端和 Bookie 中的 LAC 都为-1在一轮写入后Bookie 中的 LAC 就会滞后于客户端了好在这个问题并不会影响到一致性协议的正确性下文中会提到。
故障处理
因为需要容忍一定数据的节点故障所以一致性协议复杂的部分都在故障处理逻辑。接下来我们先看写入失败场景。
写入失败
Kafka、Raft 等集成一致性协议部分存储节点异常处理相对简单如果是 Leader 节点故障切换到其他节点进行读写即可。如果是 Follower 节点异常通常不需要做任何操作只需要等节点恢复后从 Leader 节点同步数据补齐差异。由于节点是带存储的所以可以容忍较长时间的节点故障。而 BK 是在客户端实现的一致性协议客户端不带存储没写成功的数据需要缓存在内存里显然可缓存的数据是非常有限的同时没有写成功 LAC 是不能推进的整个写入也就停止了。所以 BK 在写入失败时最好是方式就是挑选新的 Bookie 节点重新写入创建 Ledger 时已经指定了初始的 Bookie 列表后续的 Bookie 列表变更都称为 EnsembleChange。
EnsembleChange
如下图所示写入失败时会替换新节点重新写入同时也会把新元数据更新到 ZK。每次 EnsembleChangeEnsembles 中就会新增一个 Fragment起始位置为当前 LAC1。 写入存储节点失败简单可以分两种情况如果还没有收到Qa 个数成功响应前收到了错误响应比如超时等会立即执行 EnsembleChange。如果已经收到了 Qa 个数成功响应更新 LAC后续的错误响应只会记录下失败 Bookie 节点在下一次写入时再触发 EnsembleChange。如果是不可恢复的异常会直接返回写入失败不会做 EnsembleChange。 EnsembleChange 和初始创建 Ledger 一样需要先选择新的 Bookie 节点列表替换掉失败的 Bookie 节点列表然后把替换后的元数据更新回 ZK代码如下: 更新完元数据后重写数据到新节点或者根据条件重新触发 EnsembleChange代码如下: 简而言之在写入 Bookie 异常时BK 客户端都会尝试切换 Bookie 节点重写数据。如果遇到无法恢复的状态码就会直接往外层抛出异常。上层使用方比如 Pulsar 侧接收到这个异常后会正常关闭当前 Ledger然后创建新的 Ledger 继续写入。
客户端故障
如果存储 Bookie 节点故障客户端可以切换其他 Bookie 继续完成写入。接下来我们看看如果正在写入的客户端故障了应该怎么处理。在 Pulsar 架构里BK 客户端是 Pulsar Broker 内的一个组件当 Pulsar Broker 故障后分区的 Leader 会切换到新 Broker 继续服务。由于 BK 客户端故障时没有来得及正确关闭 Ledger。当前 Ledger 数据是不可读的因为 LAC 信息已经丢失。所以新 Broker 主要任务就是要恢复 LAC 信息然后正确关闭 Ledger这样 Ledger 内数据就可以正确读取。最后再创建新的 Ledger 来恢复写入。由于恢复过程对应分区无法写入所以要求恢复过程越快越好不可存在长时间阻塞。
Fence Bookie
在恢复之前由于老 Broker 有可能是短暂假死(比如长时间 gc、网络隔离等)后续可能还会持续向当前 Ledger 写入数据导致脑裂。所以恢复的第一步就是要通知对应 Bookie 节点后续禁止往这个 ledger 里面写任何数据这个过程称为 Fence。过程相对简单就是向所有 Essemble 中的 Bookie 节点发送请求让其禁止该 Ledger 的后续写入。如下图所示: 由于当前 Bookie 节点不能保证全部存活同时需要满足快速恢复所以需要考虑客户端至少收到多少个成功响应才能认为 Fence 操作执行成功。这里可以反过来想一下客户端每次需要写入 Qw 个节点然后收到 Qa 个成功响应就能认为写入成功我们只需要让其凑不足 Qa 个成功响应就好了也就是我们 Fence 掉的节点数只要大于 Qw - Qa那写入客户端就一定凑不足 Qa 个成功响应我们暂且称这个数量为 Qf。例子如下
Qw 5, Qa 3 Qf 5 - 3 1 3
Qw 3, Qa 2 Qf 3 - 2 1 2
Qw 3, Qa 3 Qf 3 - 3 1 1 // QwQa 场景需要fence的节点数最少恢复最快由于每次写入都要从 E 个 Bookie 中挑选出 Qw 个节点来条带化写入所以还需要保证任意一组 Qw我们都 Fence 掉了对应的 Qf 个节点。例子如下
假设 E3,QwQa2, Bookies列表为0,1,2, Qf2-211
由于会条带化写入每次写入挑选两台节点全部组合为:
0,1 // 0,1节点至少要成功fence 掉一台
1,2 // 1,2节点至少要成功fence 掉一台
2,0 // 2,0节点至少要成功fence 掉一台
易得 0,1,2 三台节点节点至少要成功fence 掉两台
同理可证如果按上文中推荐配置 EQwQa, 任何情况下都只需要fence掉任意1台节点即可fence 节点数量最少恢复最快。实际 BK 中实现的代码逻辑是统计任意 writeSet 未成功响应数大于等于 Qa 都认为未成功功能一致代码如下: 恢复 LAC
恢复主要的任务就是恢复 LAC 信息上文中已经介绍 LAC 信息位于每条数据中存储在 Bookie 里面。接下来看看恢复整个过程。
修改元数据状态
第一步: 把 Ledger 状态改成 Recovery 状态。 读取初始 LAC
第二步: 读取 Bookie 中的 LAC下图代码中可以看到 Fence Bookie 操作其实与查询最后 LAC 为同一个请求只是携带了 Fence 标识。 取最大的 Bookie LAC 值返回。 初始 LAC maxzkLac, bookieLac作为初始 LAC。 恢复真实 LAC
由于 Bookie 中存储的 LAC 有滞后性也就是真实 LAC故障前内存中的LAC往往大于目前从 Bookie 查询到的 LAC 最大值。所以我们还需要恢复真实的 LAC在找到真实的 LAC之前我们先确定一个基本的原则。最终恢复的 LAC 可以大于真实 LAC但是绝对不能小于真实 LAC会导致数据丢失。比如一条数据已经成功写入到 Qa 个节点中但是客户端还没来得及接受到 Qa 个成功响应不更新 LAC就故障了那么恢复时把这条数据 EntryID 更新到 LAC 也是合理的。换句话说当数据成功写入到了 Qa 节点中的那一刻真实 LAC 就应该立即更新只不过客户端内存中的 LAC 也存在一定的滞后需等收到响应。所以恢复过程可以以当前读取到的 LAC 作为起始 LAC, 依次往后面查询下一条数据LAC 1如果这条数据已经存在到 Qa 个节点中那这条数据就是可恢复的向前推进 LAC 并查询下一条数据。直到某条数据存在的节点数少于 Qa , 就可以认定这条数据是不可恢复的。那当前的 LAC 就是真实的 LAC, 恢复过程结束。以上方案判断某条数据是否可恢复是根据收到的存在响应大于等于 Qa所以对于每条数据都需要查询全部节点。BK 客户端实际上使用另一种更加快速的方式来判断数据是否可恢复。数据是否可恢复判断方式如下
串行发送读请求如果收到存在响应认定数据可恢复推进 LAC, 继续恢复下一条数据。其余响应依次读取下一个 Bookie 存储节点如果已经有 Qf Fence 章节中定义个节点明确返回不存在该数据表明该数据不可能存在于 Qa 个节点上后续的数据也不用查询了恢复结束当前 LAC 为最终需要恢复的 LAC。
下面使用伪代码描述如果判断单条数据是否可恢复。
int notFoundCount 0;
for(Bookie bookie : writeSet) { result bookie.query(lac 1); switch( result) { case : found return 数据可恢复; // 可恢复推进LAC 读取下一条数据 case : not foundif( notFoundCount Qf) { // 终止恢复当前 LAC 作为最终 LAC return 数据无法恢复; }; break; case : unkonwn // 连接不上、超时等 }
}
throw Exception(多节点未知终止恢复) // 无法恢复等待重试举例如下:
Qw3, Qa 2, Qf 3 - 2 1 2
向bookie0, bookie1, bookie2 同时发出读取请求1. 收到 bookie0 或 bookie1 或 bookie2 存在响应可恢复
2. bookie0 否定回答bookie1 否定回答 bookie2 未知回答不可恢复
3. bookie0 否定回答bookie1 未知回答 bookie2 未知回答恢复失败需要重试需要特别注意由于 LAC 非常重要如果最终恢复的 LAC 小于实际 LAC就会发生日志截断相当于这段数据就丢失了。所以我们恢复过程中需要明确收到 Qf 个Bookie 节点返回数据不存在而未知返回节点暂时不可用超时等不能等同于数据不存在。
以上方案为了快速恢复只需要任意一台节点返回数据存在就认为可恢复从而推进 LAC。假设这条数据实际上存在份数不足 Qa就不能容忍后续对应的存储节点故障了。所以会把这条数据覆盖写入全部的存储节点写成功后自然 LAC 会更新写入失败未收到 Qa 个成功会导致恢复失败。Bookie 是 KV 存储支持幂等覆盖写。
代码如下:
开始执行恢复 LAC。 发起读请求。 返回失败后向下一个节点发送读请求如果收到 Qf 个不存在响应恢复终止。 读取成功包含失败重试其他节点: 回写这条数据回 BookieLAC自然推进 触发下一条数据恢复 读取失败Qf 个节点明确返回数据不存在: 当前 LAC 就为需要恢复的真正 LAC恢复成功。 消费
消费逻辑相对简单当前 Ledger 在写入时只有当前写入的客户端可以读取数据使用内存中的 LAC 防止读取越线。其他客户端只能读取已关闭的 Ledger首先从 ZK 中获取元数据包含 LAC然后正常向对应 Bookie 发起请求即可。失败后尝试下一个 Bookie 节点直到成功或者所有节点都尝试过。恢复 LAC 的读取过程和正常读取数据使用同一套逻辑且相对简单这里就不做代码分析。
可以看到 BK 客户端读取消息是按单条消息来读取的会造成请求数较多。高版本 BK 已经做了一定优化客户端提供了批读能力可以和服务端一次交互就读到多条消息。这里有个前提条件就是数据不能条带化写入因为条带化写入会让数据分散到多台节点单台节点内数据不连续所以生产环境还是建议配置 EQwQa 。
结语
本文主要从客户端视角分析 BK 一致性协议设计理念以及实现原理包含 Leger 创建、数据的写入、读取以及客户端服务端故障恢复等内容。可以看到 BK 的一致性协议还是有一些有趣的地方并且实实在在的解决了一些问题也能理解到 Pulsar 的存算分离并不是简单的加一层无状态代理层来实现的。
