HBase 健康检查

记录最近工作中遇到的一个需求：

• IDC 容灾，HBase 部署在两个 IDC 上，且各 IDC 独立运行，需要保证数据 写 到两边的 HBase 中。
• 分区容错性：当任一 IDC 不可用时系统可以正常提供服务。
• 低一致性，可以因容忍程序异常导致的少量数据不一致的情况。

最终敲定的方案为 应用端 做 双写，该方案的关键点在于如何保证在 HBase 故障时应用正常运行。 为此，开发了一个 HBase 健康检查程序，通过 周期性地 检查 某张表是否存在 确定 HBase 是否可用。

项目中使用 asynchbase 库作为 HBase 客户端，部分示例代码如下：

public class HealthyChecker implements AutoCloseable {

	private final HealthyCheckerConfig config;

	private final HBaseClient hbaseClient;

	private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

	private final AtomicLong unhealthyCount = new AtomicLong(0);
	
	// ...
	
	public void start() {
		executor.scheduleWithFixedDelay(
			this::check,
			0,
			config.checkInterval().toMillis(),
			TimeUnit.MILLISECONDS);
	}		
		
	void check() {
		try {
			hbaseClient.ensureTableExists(config.targetTableName())
				.join(config.timeoutMillis());
		} catch (IllegalArgumentException e) {
			// Never reach here.
			throw e;
		} catch (InterruptedException e) {
			log.warn("Interrupted while waiting for healthy check.", e);
			Thread.currentThread().interrupt();
		} catch (Exception e) {
			long currentCount = unhealthyCount.incrementAndGet();
			log.error("Checked hbase is unhealthy. Current unhealthy count is {}, max tolerance is {}",
				currentCount, config.maxTolerance(), e);
		}

		// reset unhealthy count
		unhealthyCount.set(0);
	}
	
	public boolean isHBaseAvailable() {
		return unhealthyCount.get() <= config.maxTolerance();
	}
	
    // ...
}

该例子的完整代码可以参考：https://github.com/zou-can/java-demos/tree/main/async-hbase 。

关于 asynchbase 库

该库是 OpenTSDB 组织开发的一个异步 HBase 客户端， 在 HBase 早期的版本可能存在读写效率低的问题，因此有的这个库。（个人猜测）

WARNING

asynchbase 库于 2018 年 5 月 20 日发布最后一个版本后就 停止 更新了， 相关的使用方法和文档也比较少，如有可能，还是建议使用 官方 的库进行开发。

下面介绍几个使用方面的关键点。

Deffered<Object>

该库中对 HBase 的所有操作都会返回一个 Deffered<Object> 类型的对象。 据该类的 java doc 中称，该类的设计基于 Twisted Deffered Python API。 该类在使用上有点像 CompletableFuture， 但一个很重要的区别是：Deffered 执行回调（callback）的线程为创建它的线程，Deffered 不会 创建和管理新的线程或线程池。

以下内容来自 Deffered 类的文档注释：

The thread that executes the callback chain is the thread that hands the initial result to the {@code Deferred}. This class does not create or manage any thread or executor.

Deffered<Object>#join()

该库会 缓存 对 HBase 的请求，并定时刷新缓冲区。 调用 Deffered#join 方法会等待刷写缓冲区，其时间开销与 hbase.rpcs.buffered_flush_interval 参数有关，默认值为 1000 ms。

PleaseThrottleException

这个异常翻译成中文是“请限流异常”，意味着应用应进端行 限流 处理。 客户端请求都是异步非阻塞的，因此程序可能会以高于 HBase 能够处理的速率发送请求。 当请求数超过配置的阈值时，会抛出该异常。

一个典型的情况是，当 HBase 进行 Region 切分时，Region server 会短暂下线， 但客户端会缓存 Region 信息，此时请求可能无法正确处理。 客户端会将处理失败请求缓存到名为 NSRE 的队列中，若队列已满，则会抛出 PleaseThrottleException 异常，直到 Region 重新上线。

因此，应用端代码应捕获并恰当处理该异常。

更多关于如何处理该异常的建议可以参考 OpenTSDB Github -- issue-162 。