Hbase replication源码分析

作者介绍：熊训德（英文名：Sundy），16年毕业于四川大学大学并加入腾讯。目前在腾讯云从事 hadoop 生态相关的云存储和计算等后台开发，喜欢并专注于研究大数据、虚拟化和人工智能等相关技术。

Replication核心原理

hbase.replication.source.service 配置说明了使用哪个类来作为 Hbase 的 Replication 实现类。

hbase.replication.source.service 这个默认配置实现就是

org.apache.hadoop.hbase.replication.regionserver.Replication

其中 Replicaton 的类图如下：

最重要的几个成员在类图中列出来了，

Replication 类在 HRegionServer 中有一个 ReplicationService 类型的引用，即是 replicaitonSourceHandler，RegionServer 以这个引用来控制集群的复制；

Replication 类在复制和管理方面比较重要的几个类是：ReplicationManager，ReplicationTracker 和 ReplicationLoad。

其中 ReplicationManager 主要管理 Replication，包括线程池的初始化、结束，以及通过 ZK 管理 RS 的加入和可能的异常（failover）。在 ReplicationManager 中比较重要类的是 ReplicationSource，这个类是实际的 Replication 执行者，在 HRegionServer 对 Replication 初始化时即被启动，会一直去扫描其在 Znode 中对应的 HLog 文件，有新的记录就往 peers(Slave cluster) 中传输，并记录到达了的对应 peer 相应传输点。

ReplicationTracker 主要用于跟踪 Zookeeper 中 /hbase/replication 中 znode 的状态，Hbase 的 Replication 是通过 zk 来协助完成的，zk 中有记录每个 wal 文件（HLog）传输的 checkpoint，以及当某个 rs 挂了后也需要 zk 来协助把原来 rs 中负责传输的 wal 分配到另一个 wal 上去。默认的实现类是 ReplicationTrackerZKImpl。

ReplicationLoad 是用于记录 Replication 过程中的参数状态（传输队列长度，时延等），通过一个线程池（SchdulExecutorpool）定时的获取这些状态来实现，可以通过 HbaseMetric 以 JMX 方式获得或者在 WebUI 的对应RegionServer 中也能查看。用于监控 Hbase Replication。

Replication 的初始化是在HRegionServer初始化完成，开始向 HMaster Report 的过程（handleReportForDutyResponse()）中开始：它在 HRegionServer 中有 replicaitonSourceHandler 引用。在 handleReportForDutyResponse()方法中调用 setupWALAndReplication 实例化和初始化 Replication 。在初始化过程中，包括 ReplicationLoad (以及对应的监控线程池)，ReplicationTacker 和 ReplicationManager 都要进行初始化。而且Replication实例还会被注册为 WALActionLitener，当 LogRoller 时会回调对应方法。

在写 wal 也注册了 listener，但是 Replication 并未做传输的动作。很可能是考虑到了性能方面的影响，Hbase 的Replication 方案是异步传输。具体做法是初始化后 ReplicationManger 中对应的 ReplicationSource 实例会去扫描 HLog 文件去传输：

上图是 client 端 put/delete 时，replication 的时序图。

1.client 把 KeyValue 提交到 RS 后封装成 WALEdit 写到 WAL，成功后（fush()）即返回。

2.这时 RS 中有一个异步的后台 ReplicationSource 线程会异步的去扫描 HLog（包括正在写的和已经滚动了的 HLog），把新增的 WAL 以 RPC 调用的方式发送到对应的 Slave Cluster 。

3.写成功后会到 ZK 上更新日志对应成功的 checkpoint 。

图中红色框中的即是 ReplicationSource 线程所在类。RSSource 即是 SourceCluster 的 RegionServer ，RSSink 是 SlaveCluster 的 RegionServer 。

复制的核心过程

这里重点解析 ReplicationResource 类，与之相关的重点类图如下：

在 replication 过程有两个关键的 Queue：

1.第一个在 ReplicationManager 和 Replication 都有的一个引用 ReplicationQueues，这个用于控制整个 RegionServer 的复制队列（replication queues）。具体实现类是 ReplicationQueuesZKImpl，因为 replication 可能会有多个 Slave Cluster 而且复制过程很可能会挂掉，所以 Hbase 通过 Zookeeper 的协助来完成 replication 以及 failover 过程。与 replication 相关 znode 有如下结构：

其根 znode 是 {zookeeper.znode.parent}/replication (图中是默认值 /hbase/replication)，在其子节点中有相关的 peer 的状态以及分配的 table ，另一块子节点记录了不同的各个 RS 上所传输的 HLog 队列以及对应的 checkpoint 。

这个 queue 的在三种情况下，在对应的 rs 的 peer 子 znode 下会添加文件：

1.addSource 时，rs 启动时把相应 HLog 文件名

2.preLogRoll 时，把新滚动的 HLog 添加上

3.FailOver 时，把挂了的 rs 的 HLog 文件名和 checkpoint 移动到相应 rs 上

2.另一个 queue 是在 ReplicationSource 中用 hdfs Path 来代表具体的 path 的 queue 。具体作用是记录 Hlog 的Path，通过这个类打开 Reader 读取 WALEdit 。

其实现类是 PriorityBlockingQueue<Path>，这个类首先是线程安全类，其实是基于数组堆的优先队列。所含对象的排序不是 FIFO ,而是依据对象的自然排序顺序或者是构造函数的 Comparator 决定的顺序。在其构造函数中的 Comparator 是 LogsComparator 。

这个 LogsComparator 并不复杂，就是基于开始时间实现了 Compare 接口的类，这样使用这个优先队列时间最早（小） Path 的排在最前面。

其中最开始时间用 getTS()方法从 wal file 名字中获得。即是把 Wal file 的名字以“.”（点号）划分，然后最后一段就是开始时间（一个时间戳），比如一个 wal file path 是这样的

hdfs:///hbase/WALs/172.16.0.148,6002,1490267303489/172.16.0.148%2C6002%2C1490267303489.default.1490626004532

对应 HLog 的名字是：172.16.0.148%2C6002%2C1490267303489.default.1490626004532

对应 HLog 开始时间戳即是：1490626004532

这个 queue 增加有三种情况：

1.addSource

2.postLogRoll

3.RsFailover

在 Master Cluster 被源源不断的写数据过程中，其 WAL 文件（Hlog）的文件名被加到 ZK 对应的 znode 下，对应为ReplicationQueues，与此同时对应 Hlog 的 Path 则被加载到内存中对应为 PriorityBlockingQueue<Path>。ReplicationSource持有这两个队列的引用，在其run中不断的对这些Hlog传输到配置的不同的Slave Cluster中，具体 replicate 的时序图如下：

1.ReplicationSource 的异步线程从内存的队列中取出当前需要处理的 HLog 的 Path 并打开它。这时候

2.以 WALEdit 为单位，从上次标记的 checkpoint 开始( zk 中记录)扫描这个 HLog，和写 wal 时类似把 WALEdit 和 WALKey 重新封装成 WAL.Entry，每个封装后的 Entry 按照 wal 中顺序组成一个List<Entry>。扫描过程一直到出现了下面几种情况才会终止：

a.entry 的数量超过了配置 replication.source.nb.capacityr，默认是25000

b.所有 entry 的总大小超过了配置 eplication.source.size.capacity，默认是64M

c.碰到了包括 EOF 在内的等各种异常

3.如果是前两种情况会调用 shipEdit 开始往 slave cluster 发送，如果碰到了异常，如果是 eof 异常则也会发送，如果是其他的异常会 sleep 后等待下一次唤醒。

4.在 shipEdit 方法中，会调用 replicate 默认实现类 HbaseInterClusterReplicatoinEndpoint 的 replicate 方法，在这个方法中实际操作 replicate 一队 entry(List<Entry>)。

a.replication 有一个流量控制的配置，受两方面影响一个是流量，另一个是周期（cycle）；流量的控制通过计算一段时间内发送的总的 entry 大小来判断，cycle 通过上一次发送时间和当前时间来对比。通过这两个参数确定当前是否需要 sleep 的间隔。过了这个间隔后即被唤醒。可配置不开启流量控制。

b.为了节省内存及时的 gc，重新封装List<Entry> 成 ReplicationContext 作为发送给 slave cluster 的单元。

c.随机选择一个 Slave Cluster 的 Rs 作为本次发送的目的端（如果 slave cluster Zk 监测的变更使用上一次的 RS），选择的规则是根据配置 replication.source.ratio，默认是0.1，这个配置表示 master 集群从 slave 集群中选择 rs 发送的比例，可以调节 slave 集群的写压力。

d.通过 AdminService 的 rpc 接口把这批 entry 发送到选定的 RS 上。发送的具体：把 entries 根据不同的表分开再使用 Table.batch 接口发送到 RS 中，这里是有重试机制，所以在 batch 前使用 master 和 slave 的 UUID 作为 key ，防止重发。

5.设置 HLog 新的 checkpoint 或者清理旧的 HLog

6.全部 entries 发送成功后，更新相应的监控数据，包括本批次数据的传输时延(Replication Lag)，本批次返回的时间戳(TimeStampsOfLastShippedOp)，最后一个传输的时间代(AgeOfLastShippedOp:当前时间-List <entry>中最后一个 entry 插入 WAL 的时间)。

failover 过程

Replication 的 Failover 主要是通过 zk 协助实现的。

1.如果是 Slave Cluster 的某个 Rs 挂了，Master 在 replicate()中有重试，包括在 table.batch、ReplicationEndPoint 和 ReplicationSource 都是有重试的，当重试超过一定次数后，把异常一直抛到 ReplicationSource 这层后会把这个 RS 加入到 deadServer 中，后续在上一步4.c 选择 rs 时不会再选择这个 RS。

2.如果是 Master Cluster 的某个 RS 挂了，其他的 RS 中 ReplicationQueues 都有会 watch 到，它们会去 lock 住挂了的 rs 的 znode，因为是排他的，只有一个 RS 能够抢到并把这些未处理完的 Hlog 放到自己的 znode 目录下依序处理。

比如最开始是这样子的，然后223给挂了

那么 rs/223就会被 lock 住，然后再把其下 znode 都拷贝到自己 znode 目录下，并把原来的 znode 给删了，然后会去根据其 hlog 所在 checkpoint 开始去 replicate 给对应 slave 集群：

其具体的时序图如下：


所有 RS 中 ReplicationManager 的 ReplicationTrackerZkImpl 都会去跟踪(注册)/hbase/rs 目录下节点的 NodeDelete 事件，因为这个目录的先对应 rs 都是临时的（ephemeral），一旦某个 rs 挂了对于会话就会关闭，znode 也会被删除，那么注册在这个 znode 上的 ZKWatcher 的 nodeDelte 的 Listener 将被调用。这个回调最终会在 ReplicationSourceManager 中启动一个线程独立去完成 failover 任务。

如上所述，方法即是：

a.先 lock 住/hbase/replication/rs 对应 rs 下的目录，其他的 rs 发现已经被 lock 后就会直接退出线程。

b.把原来的目录拷到自己 d 的/hbase/replication/rs 目录下，并改其 peer 的名字，改名规则是：peer 名称+原来 ClusterId 名称。改名的原因是可以让 replication 处理时分辨出是 recover 后的 Rs 的 WAL，会优先迅速的处理掉。

c.把原来的 rs 目录删除。

因为保存在 znode 中的 Hlog 有保存相应的 checkpoint，所以即是挂了，新的 rs 也可以从上次 Hlog 的 checkpoint 开始往 slave 集群中发送数据。

附：WALEdit 生命周期

1.client put/delete RPC 发到 Hbase

2.RegionServer 把相应的请求写到 WAL

3.假如改变的 cell 对应的 column family 的范围有在 replication 中（支持到 column family），这个 wal edit 将增加到 replication 对应的 queue 中。(这个加的过程是异步的，先在 listener 中 Replication.visitLogEntryBeforeWrite()，Replication 即是一个 WALActionListener，在后面扫描中才加入像4中所叙述。)

alter 'your_table', {NAME => 'family_name', REPLICATION_SCOPE => '1'}

4.在一个独立的线程中，作为批处理的一部分，edit 从 Hlog 读取。

5.edit 加上主集群的 UUD（重新封装），然后加入到一个 buffer(？entriesList<Entry>)。当这个 buffer 满了或者 reader 到达了文件的底部， buffer 会发送给一个 slave 集群随机的一个 RegionServer。

6.RegionServer 顺序读取 edit 并把 edit 分离到各个buffer 中，每一个表一个 buffer 。（这个 buffer 和5中 buffer 一样？是一样）在所有的 edit 都被 read 后，每一个 buffer 会使用org.apache.hadoop.hbase.client.Table 进行 flush。为了防止循环复制 Master 集群的 UUID 和 slave 集群的 UUIDs 消耗了数据后就会保存了这些 edit。

7.在 master 集群中，WAL 的当前复制的 offset 将被保存在 Zookeeper 中。

8.前三部的Edit插入步骤都是一样的

9.又在一个单独的线程中，RegionServer 读取(reads)、过滤(filters)和编辑（edits）HLog edit用相同的方式就像上面一样。slave Regionserver 不回应RPC调用。

10.master 集群将 sleep 和尝试一个可配置的次数。

11.如果 slave 集群的 RegionServer 仍然不可用，master集群将选择一个新的 RegionServer 并且重新发送 edit 的 buffer 。

12.与此同事，WAL 的回滚日志也将加入 Zookeeper 的 replication 对应队列。日志将归档到各自的 RegionServer 中，方式是通过移动 wal 文件目录到一个集中的日志目录。日志将在内存中更新它们的目录在 replication 的线程中。

13.当 slave 集群最终可用时，master 集群中的 buffer 将用同样的方式处理。master 集群会将积压的日志都复制到停机期间的 slave 集群。