1.3 被遗弃的hbase客户端使用代码
被遗弃的创建方式一：直接通过HTable(Configuration conf, final String tableName)创建
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.client.write.buffer", "2097152");
configuration.set("hbase.zookeeper.quorum","192.168.199.31,192.168.199.32,192.168.199.33,192.168.199.34,192.168.199.35");
Table table = new HTable(configuration, "tableName");
//3177不是我杜撰的，是2*hbase.client.write.buffer/put.heapSize()计算出来的
int bestBathPutSize = 3177;
try {
// Use the table as needed, for a single operation and a single thread
// construct List<Put> putLists
List<Put> putLists = new ArrayList<Put>();
for(int count=0;count<100000;count++){
Put put = new Put(rowkey.getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName1".getBytes(), "columnValue1".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName2".getBytes(), "columnValue2".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName3".getBytes(), "columnValue3".getBytes());
put.setDurability(Durability.SKIP_WAL);
putLists.add(put);
if(putLists.size()==(bestBathPutSize-1)){
//达到最佳大小值了，马上提交一把
table.put(putLists);
putLists.clear();
}
}
//剩下的未提交数据，最后做一次提交
table.put(putLists)
} finally {
table.close();
connection.close();
}
被遗弃的方式二：通过HConnectionManager.createConnection(Configuration conf)获取HTableInterface
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.client.write.buffer", "2097152");
configuration.set("hbase.zookeeper.quorum","192.168.199.31,192.168.199.32,192.168.199.33,192.168.199.34,192.168.199.35");
HConnection connection = HConnectionManager.createConnection(configuration);
HTableInterface table = connection.getTable(TableName.valueOf("tableName"));
//3177不是我杜撰的，是2*hbase.client.write.buffer/put.heapSize()计算出来的
int bestBathPutSize = 3177;
try {
// Use the table as needed, for a single operation and a single thread
// construct List<Put> putLists
List<Put> putLists = new ArrayList<Put>();
for(int count=0;count<100000;count++){
Put put = new Put(rowkey.getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName1".getBytes(), "columnValue1".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName2".getBytes(), "columnValue2".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName3".getBytes(), "columnValue3".getBytes());
put.setDurability(Durability.SKIP_WAL);
putLists.add(put);
if(putLists.size()==(bestBathPutSize-1)){
//达到最佳大小值了，马上提交一把
table.put(putLists);
putLists.clear();
}
}
//剩下的未提交数据，最后做一次提交
table.put(putLists)
} finally {
table.close();
connection.close();
}
2.hbase客户端源码解读
前面我们说过，推荐的使用hbase客户端的方式如下：
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
那源代码的查看就从这两行代码开始，先来看下ConnectionFactory.createConnection(configuration)
2.1ConnectionFactory.createConnection(Configuration conf)
先看下createConnection(Configuration conf)的源代码,如下：
public static Connection createConnection(Configuration conf) throws IOException {
return createConnection(conf, null, null);
}
传入我们构造的Configuration对象，然后调用了ConnectionFactory.createConnection(Configuration conf, ExecutorService pool, User user)，继续看ConnectionFactory.createConnection(Configuration conf, ExecutorService pool, User user)的源代码，如下：
public static Connection createConnection(Configuration conf, ExecutorService pool, User user)
throws IOException {
//因为上面传入的user为null，这里代码不会执行
if (user == null) {
UserProvider provider = UserProvider.instantiate(conf);
user = provider.getCurrent();
}
return createConnection(conf, false, pool, user);
}
这里继续调用了ConnectionFactory.createConnection(final Configuration conf, final boolean managed, final ExecutorService pool, final User user)，那么我们继续看下相关代码，如下：
static Connection createConnection(final Configuration conf, final boolean managed, final ExecutorService pool, final User user)
throws IOException {
//默认HBASE_CLIENT_CONNECTION_IMPL = "hbase.client.connection.impl"
//hbase.client.connection.impl供hbase使用者实现自己的hbase链接实现类并配置进来使用
//默认hbase已经提供了实现，无需实现，那么这里就取默认实现ConnectionManager.HConnectionImplementation.class.getName()
//默认hbase的connection实现类也即HConnectionImplementation类
String className = conf.get(HConnection.HBASE_CLIENT_CONNECTION_IMPL,ConnectionManager.HConnectionImplementation.class.getName());
Class<> clazz = null;
try {
clazz = Class.forName(className);
} catch (ClassNotFoundException e) {
throw new IOException(e);
}
try {
// Default HCM#HCI is not accessible; make it so before invoking.
//这里调用HConnectionImplementation类的构造方法HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user)
Constructor<> constructor = clazz.getDeclaredConstructor(Configuration.class, boolean.class, ExecutorService.class, User.class);
constructor.setAccessible(true);
return (Connection) constructor.newInstance(conf, managed, pool, user);
} catch (Exception e) {
throw new IOException(e);
}
}
}
上面的代码默认调用ConnectionManager.HConnectionImplementation类返回Connection对象，继续跟踪HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user)代码：
HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user) throws IOException {
//这里代码我们需要重点关注
this(conf);
//这里this.user=null
this.user = user;
//这里this.batchPool=null
this.batchPool = pool;
//这里this.managed=false
this.managed = managed;
//这里setupRegistry()默认从hbase.client.registry.impl获取客户端使用者实现的zookeeper注册类，没有配置就默认创建ZooKeeperRegistry类对象并设置，这个类非常重要，客户端与zookeeper的交互类就由此类负责
this.registry = setupRegistry();
//默认通过ZooKeeperRegistry对象从zookeeper获取hbase集群的clusterId
retrieveClusterId();
//如果Configuration没配置hbase.rpc.client.impl就默认创建RpcClientImpl并设置给this.rpcClient
this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);
this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
// Do we publish the status
//如果Configuration没配置hbase.status.published就默认设置shouldListen=false
boolean shouldListen = conf.getBoolean(HConstants.STATUS_PUBLISHED, HConstants.STATUS_PUBLISHED_DEFAULT);
//如果Configuration没配置hbase.status.listener.class就默认创建MulticastListener对象并设置给listenerClass
Class< extends ClusterStatusListener.Listener> listenerClass = conf.getClass(ClusterStatusListener.STATUS_LISTENER_CLASS, ClusterStatusListener.DEFAULT_STATUS_LISTENER_CLASS, ClusterStatusListener.Listener.class);
if (shouldListen) {
if (listenerClass == null) {
LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " + ClusterStatusListener.STATUS_LISTENER_CLASS + " is not set - not listening status");
} else {
//这里通过hbase事件监听器监视hbase服务端事件，当hbase服务端服务不可用时，调用rpcClient.cancelConnections关闭链接
clusterStatusListener = new ClusterStatusListener(
new ClusterStatusListener.DeadServerHandler() {
@Override
public void newDead(ServerName sn) {
clearCaches(sn);
rpcClient.cancelConnections(sn);
}
}, conf, listenerClass);
}
}
}
上面的代码我们主要关注this(conf);另外一个需要注意的就是方法setupRegistry()，setupRegistry()这里默认设置的是org.apache.hadoop.hbase.client.ZooKeeperRegistry，这一行并将在后面继续分析，其它的代码都比较简单，我在上面代码中已经做代码注释，继续看this(conf)代码：
protected HConnectionImplementation(Configuration conf) {
//这里把客户端使用者传入的Configuration赋值给this.conf
this.conf = conf;
//这里HConnectionImplementation基于我们传入的Configuration构建了自己的Configuration类对象this.connectionConfig
this.connectionConfig = new ConnectionConfiguration(conf);
this.closed = false;
//客户端使用者的Configuration没有配置hbase.client.pause，那么就设置默认值this.pause=100
this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE, HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
//客户端使用者的Configuration没有配置hbase.meta.replicas.use，那么就设置默认值this.useMetaReplicas=false
this.useMetaReplicas = conf.getBoolean(HConstants.USE_META_REPLICAS, HConstants.DEFAULT_USE_META_REPLICAS);
//从this.connectionConfig里获取值设置，而客户端使用者的Configuration没有配置hbase.client.retries.number就默认设置this.numTries=31
this.numTries = connectionConfig.getRetriesNumber();
//客户端使用者的Configuration没有配置hbase.rpc.timeout，那么就设置默认值this.rpcTimeout=60000毫秒
this.rpcTimeout = conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
if (conf.getBoolean(CLIENT_NONCES_ENABLED_KEY, true)) {
synchronized (nonceGeneratorCreateLock) {
if (ConnectionManager.nonceGenerator == null) {
ConnectionManager.nonceGenerator = new PerClientRandomNonceGenerator();
}
this.nonceGenerator = ConnectionManager.nonceGenerator;
}
} else {
this.nonceGenerator = new NoNonceGenerator();
}
//跟踪region的统计信息
stats = ServerStatisticTracker.create(conf);
//hbase客户端异步操作类
this.asyncProcess = createAsyncProcess(this.conf);
this.interceptor = (new RetryingCallerInterceptorFactory(conf)).build();
this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf, interceptor, this.stats);
this.backoffPolicy = ClientBackoffPolicyFactory.create(conf);
if (conf.getBoolean(CLIENT_SIDE_METRICS_ENABLED_KEY, false)) {
this.metrics = new MetricsConnection(this);
} else {
this.metrics = null;
}
this.hostnamesCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
this.metaCache = new MetaCache(this.metrics);
}
上面代码比较重要的一点是，尽管客户端传入了Configuration，但是HConnectionImplementation不会直接使用客户端传入的Configuration，而是基于客户端传入的Configuration构建了自己的Configuration对象，原因是客户端传入的Configuration对象只给了部分值，很多其它值都未给出，那么HConnectionImplementation就有必要创建自己的Configuration，首先构建自己默认的Configuration，然后把客户端已经设置的Configuration的相关值覆盖那些默认值，客户端没设置的值就使用默认值，我们继续看下this.connectionConfig = new ConnectionConfiguration(conf)的源代码：
ConnectionConfiguration(Configuration conf) {
//客户端的Configuration没有配置hbase.client.pause，那么就设置默认值this.writeBufferSize=2097152
this.writeBufferSize = conf.getLong(WRITE_BUFFER_SIZE_KEY, WRITE_BUFFER_SIZE_DEFAULT);
//客户端的Configuration没有配置hbase.client.write.buffer，那么就设置默认值this.metaOperationTimeout=1200000
this.metaOperationTimeout = conf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
//客户端的Configuration没有配置hbase.client.meta.operation.timeout，那么就设置默认值this.operationTimeout=1200000
this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
//客户端的Configuration没有配置hbase.client.operation.timeout，那么就设置默认值this.scannerCaching=Integer.MAX_VALUE
this.scannerCaching = conf.getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
//客户端的Configuration没有配置hbase.client.scanner.max.result.size，那么就设置默认值this.scannerMaxResultSize=2 * 1024 * 1024
this.scannerMaxResultSize = conf.getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
//客户端的Configuration没有配置hbase.client.primaryCallTimeout.get，那么就设置默认值this.primaryCallTimeoutMicroSecond=10000
this.primaryCallTimeoutMicroSecond = conf.getInt("hbase.client.primaryCallTimeout.get", 10000); // 10000ms
//客户端的Configuration没有配置hbase.client.replicaCallTimeout.scan，那么就设置默认值this.replicaCallTimeoutMicroSecondScan=1000000
this.replicaCallTimeoutMicroSecondScan = conf.getInt("hbase.client.replicaCallTimeout.scan", 1000000); // 1000000ms
//客户端的Configuration没有配置hbase.client.retries.number，那么就设置默认值this.retries=31
this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
//客户端的Configuration没有配置hbase.client.keyvalue.maxsize，那么就设置默认值this.maxKeyValueSize=-1
this.maxKeyValueSize = conf.getInt(MAX_KEYVALUE_SIZE_KEY, MAX_KEYVALUE_SIZE_DEFAULT);
}
上面的代码主要是初始化HConnectionImplementation自己的Configuration类型属性this.connectionConfig，默认客户端不设置属性值，这里创建的this.connectionConfig就使用默认值，这里将hbase客户端默认值抽取如下：
- hbase.client.write.buffer 默认2097152Byte，也即2MB
- hbase.client.meta.operation.timeout 默认1200000毫秒
- hbase.client.operation.timeout 默认1200000毫秒
- hbase.client.scanner.caching 默认Integer.MAX_VALUE
- hbase.client.scanner.max.result.size 默认2MB
- hbase.client.primaryCallTimeout.get 默认10000毫秒
- hbase.client.replicaCallTimeout.scan 默认1000000毫秒
- hbase.client.retries.number 默认31次
- hbase.client.keyvalue.maxsize 默认-1，不限制
- hbase.client.ipc.pool.type
- hbase.client.ipc.pool.size
- hbase.client.pause 100
- hbase.client.max.total.tasks 100
- hbase.client.max.perserver.tasks 2
- hbase.client.max.perregion.tasks 1
- hbase.client.instance.id
- hbase.client.scanner.timeout.period 60000
- hbase.client.rpc.codec
- hbase.regionserver.lease.period 被hbase.client.scanner.timeout.period代替，60000
- hbase.client.fast.fail.mode.enabled FALSE
- hbase.client.fastfail.threshold 60000
- hbase.client.fast.fail.cleanup.duration 600000
- hbase.client.fast.fail.interceptor.impl
- hbase.client.backpressure.enabled false
2.2 与zookeeper交互的ZooKeeperRegistry
上面我们分析知道客户端使用者传入的Configuration只有设置的值才会在客户端上生效，而未设置的值则交由默认值设置，另外一个非常重要的就是刚才所提到的与zookeeper交互的类org.apache.hadoop.hbase.client.ZooKeeperRegistry
package org.apache.hadoop.hbase.client;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.RegionLocations;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
import org.apache.hadoop.hbase.zookeeper.ZKTableStateClientSideReader;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.zookeeper.KeeperException;
/**
* A cluster registry that stores to zookeeper.
*/
class ZooKeeperRegistry implements Registry {
private static final Log LOG = LogFactory.getLog(ZooKeeperRegistry.class);
// hbase连接，在初始化函数中会进行设置
ConnectionManager.HConnectionImplementation hci;
@Override
public void init(Connection connection) {
if (!(connection instanceof ConnectionManager.HConnectionImplementation)) {
throw new RuntimeException("This registry depends on HConnectionImplementation");
}
//设置hbase连接
this.hci = (ConnectionManager.HConnectionImplementation)connection;
}
@Override
public RegionLocations getMetaRegionLocation() throws IOException {
//通过hbase连接中的Configuration获取zookeeper地址后，通过hbase连接获取与zookeeper交互的ZooKeeperKeepAliveConnection
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
if (LOG.isTraceEnabled()) {
LOG.trace("Looking up meta region location in ZK," + " connection=" + this);
}
//从zookeeper中获取所有的hbase region元数据信息
List<ServerName> servers = new MetaTableLocator().blockUntilAvailable(zkw, hci.rpcTimeout, hci.getConfiguration());
if (LOG.isTraceEnabled()) {
if (servers == null) {
LOG.trace("Looked up meta region location, connection=" + this + "; servers = null");
} else {
StringBuilder str = new StringBuilder();
for (ServerName s : servers) {
str.append(s.toString());
str.append(" ");
}
LOG.trace("Looked up meta region location, connection=" + this + "; servers = " + str.toString());
}
}
if (servers == null) return null;
//组装hbase RegionLocations数组进行返回
HRegionLocation[] locs = new HRegionLocation[servers.size()];
int i = 0;
for (ServerName server : servers) {
HRegionInfo h = RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO, i);
if (server == null) locs[i++] = null;
else locs[i++] = new HRegionLocation(h, server, 0);
}
return new RegionLocations(locs);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
} finally {
zkw.close();
}
}
private String clusterId = null;
@Override
public String getClusterId() {
if (this.clusterId != null) return this.clusterId;
// No synchronized here, worse case we will retrieve it twice, that's
// not an issue.
ZooKeeperKeepAliveConnection zkw = null;
try {
zkw = hci.getKeepAliveZooKeeperWatcher();
this.clusterId = ZKClusterId.readClusterIdZNode(zkw);
if (this.clusterId == null) {
LOG.info("ClusterId read in ZooKeeper is null");
}
} catch (KeeperException e) {
LOG.warn("Can't retrieve clusterId from Zookeeper", e);
} catch (IOException e) {
LOG.warn("Can't retrieve clusterId from Zookeeper", e);
} finally {
if (zkw != null) zkw.close();
}
return this.clusterId;
}
@Override
public boolean isTableOnlineState(TableName tableName, boolean enabled)
throws IOException {
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
if (enabled) {
return ZKTableStateClientSideReader.isEnabledTable(zkw, tableName);
}
return ZKTableStateClientSideReader.isDisabledTable(zkw, tableName);
} catch (KeeperException e) {
throw new IOException("Enable/Disable failed", e);
} catch (InterruptedException e) {
throw new InterruptedIOException();
} finally {
zkw.close();
}
}
@Override
public int getCurrentNrHRS() throws IOException {
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
// We go to zk rather than to master to get count of regions to avoid
// HTable having a Master dependency. See HBase-2828
return ZKUtil.getNumberOfChildren(zkw, zkw.rsZNode);
} catch (KeeperException ke) {
throw new IOException("Unexpected ZooKeeper exception", ke);
} finally {
zkw.close();
}
}
}
这个类非常重要，因为所有的与zookeeper的交互都由它来完成。
2.3HConnectionImplementation.getTable(TableName tableName)
前面我们说过，推荐的使用hbase客户端的方式如下:
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
上面2.1中已经知悉默认connection实现是HConnectionImplementation，那么这里我们继续跟踪HConnectionImplementation.getTable(TableName tableName)方法，代码如下：
public HTableInterface getTable(TableName tableName) throws IOException {
return getTable(tableName, getBatchPool());
}
继续看HConnectionImplementation.getTable(TableName tableName, ExecutorService pool)的代码：
public HTableInterface getTable(TableName tableName, ExecutorService pool) throws IOException {
//默认managed=false
if (managed) {
throw new NeedUnmanagedConnectionException();
}
return new HTable(tableName, this, connectionConfig, rpcCallerFactory, rpcControllerFactory, pool);
}
继续看HTable的构造方法HTable(TableName tableName, final ClusterConnection connection, final ConnectionConfiguration tableConfig, final RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory, final ExecutorService pool)，代码如下：
public HTable(TableName tableName, final ClusterConnection connection, final ConnectionConfiguration tableConfig, final RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory, final ExecutorService pool) throws IOException {
if (connection == null || connection.isClosed()) {
throw new IllegalArgumentException("Connection is null or closed.");
}
//设置hbase数据表名
this.tableName = tableName;
//调用close方法时，默认不关闭连接，这一点非常重要，默认调用table.close()是不会关闭之前创建的connection的，这一点在后面的table.close()里会介绍
this.cleanupConnectionOnClose = false;
//设置this.connection值为HConnectionImplementation创建的connection实现类
this.connection = connection;
//从HConnectionImplementation获取客户端传入的configuration对象
this.configuration = connection.getConfiguration();
//从HConnectionImplementation获取HConnectionImplementation基于客户端传入的configuration创建的configuration对象
this.connConfiguration = tableConfig;
//从HConnectionImplementation获取pool,HConnectionImplementation的默认pool为this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)
this.pool = pool;
if (pool == null) {
this.pool = getDefaultExecutor(this.configuration);
this.cleanupPoolOnClose = true;
} else {
//在HConnectionImplementation中已经初始化了this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)，所以这里会设置cleanupPoolOnClose，默认也不会关闭线程池
this.cleanupPoolOnClose = false;
}
this.rpcCallerFactory = rpcCallerFactory;
this.rpcControllerFactory = rpcControllerFactory;
//这个方法我们后面重点关注，其根据客户端传入的Configuration初始化HTable的参数
this.finishSetup();
}
上面的代码我已经加了注释，需要注意的是cleanupConnectionOnClose属性，该属性默认值为false，在调用table.close()方法时候，只是关闭了table而已但table后面的connection是没有关闭的，再者是属性cleanupPoolOnClose，虽然我们没有传入线程池，但是HConnectionImplementation会自己创建线程池this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)传过来使用，所以这里会设置this.cleanupPoolOnClose = false，默认在table.close()调用时候，也不会关闭线程池，那么这里这里继续跟踪上面代码最后的this.finishSetup()，代码如下：
private void finishSetup() throws IOException {
//HTable的属性connConfiguration若为空，就基于客户端传入的Configuration构建新的connConfiguration
if (connConfiguration == null) {
connConfiguration = new ConnectionConfiguration(configuration);
}
//HTable的属性设置
this.operationTimeout = tableName.isSystemTable() connConfiguration.getMetaOperationTimeout() : connConfiguration.getOperationTimeout();
this.scannerCaching = connConfiguration.getScannerCaching();
this.scannerMaxResultSize = connConfiguration.getScannerMaxResultSize();
if (this.rpcCallerFactory == null) {
this.rpcCallerFactory = connection.getNewRpcRetryingCallerFactory(configuration);
}
if (this.rpcControllerFactory == null) {
this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
}
// puts need to track errors globally due to how the APIs currently work.
//hbase的异步操作类
multiAp = this.connection.getAsyncProcess();
this.closed = false;
//hbase的region操作工具类
this.locator = new HRegionLocator(tableName, connection);
}
经过上面的分析，我们有必要看下table.close()的源代码：
public void close() throws IOException {
//如果已经关闭了，直接返回
if (this.closed) {
return;
}
//关闭前做最后一次提交
flushCommits();
//默认在构造HTable时候，cleanupPoolOnClose=false，这里不会去关闭线程池
if (cleanupPoolOnClose) {
this.pool.shutdown();
try {
boolean terminated = false;
do {
// wait until the pool has terminated
terminated = this.pool.awaitTermination(60, TimeUnit.SECONDS);
} while (!terminated);
} catch (InterruptedException e) {
this.pool.shutdownNow();
LOG.warn("waitForTermination interrupted");
}
}
//默认在构造HTable时候，cleanupConnectionOnClose=false，这里不会去关闭table持有的connection
if (cleanupConnectionOnClose) {
if (this.connection != null) {
this.connection.close();
}
}
this.closed = true;
}
2.4 HTable.put(final List<Put> puts)
我们已经通过如下代码：
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
创建了connection，其默认实现类为org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation，然后创建了table，其默认实现类为org.apache.hadoop.hbase.client.HTable，那么接下来就是分析客户端的批量提交方法：HTable.put(final List<Put> puts),代码如下：
public void put(final List<Put> puts) throws IOException {
//根据设置的缓存大小，达到缓存相关值就进行批量提交
getBufferedMutator().mutate(puts);
//不管有无数据未提交，默认autoFlush=true，那么就最后提交一次
if (autoFlush) {
flushCommits();
}
}
这里先看下HTable.getBufferedMutator()源代码：
BufferedMutator getBufferedMutator() throws IOException {
if (mutator == null) {
//从HConnectionImplementation获取pool,HConnectionImplementation的默认pool为this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)
//根据hbase.client.write.buffer设置的值，默认2MB，构造缓冲区
this.mutator = (BufferedMutatorImpl) connection.getBufferedMutator(
new BufferedMutatorParams(tableName)
.pool(pool)
.writeBufferSize(connConfiguration.getWriteBufferSize())
.maxKeyValueSize(connConfiguration.getMaxKeyValueSize())
);
}
return mutator;
}
上面的代码默认构造了一个BufferedMutatorImpl类并返回，继续跟踪BufferedMutatorImpl的方法mutate(List< extends Mutation> ms)
public void mutate(List< extends Mutation> ms) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
//如果BufferedMutatorImpl已经关闭，直接退出返回
if (closed) {
throw new IllegalStateException("Cannot put when the BufferedMutator is closed.");
}
//这里先不断循环累计提交的List<Put>记录所占的空间，放置到toAddSize
long toAddSize = 0;
for (Mutation m : ms) {
if (m instanceof Put) {
validatePut((Put) m);
}
toAddSize += m.heapSize();
}
// This behavior is highly non-intuitive... it does not protect us against
// 94-incompatible behavior, which is a timing issue because hasError, the below code
// and setter of hasError are not synchronized. Perhaps it should be removed.
if (ap.hasError()) {
//设置BufferedMutatorImpl当前记录的提交记录所占空间值为toAddSize
currentWriteBufferSize.addAndGet(toAddSize);
//把提交的记录List<Put>放置到缓存对象writeAsyncBuffer，在为提交完成前先不进行清理
writeAsyncBuffer.addAll(ms);
//这里当捕获到异常时候，再进行异常前的一次数据提交
backgroundFlushCommits(true);
} else {
//设置BufferedMutatorImpl当前记录的提交记录所占空间值为toAddSize
currentWriteBufferSize.addAndGet(toAddSize);
//把提交的记录List<Put>放置到缓存对象writeAsyncBuffer，在为提交完成前先不进行清理
writeAsyncBuffer.addAll(ms);
}
// Now try and queue what needs to be queued.
// 如果当前提交的List<Put>记录所占空间大于hbase.client.write.buffer设置的值，默认2MB，那么就马上调用backgroundFlushCommits方法
// 如果小于hbase.client.write.buffer设置的值，那么就直接退出，啥也不做
while (currentWriteBufferSize.get() > writeBufferSize) {
backgroundFlushCommits(false);
}
}
上面的代码不断循环累计提交的List<Put>记录所占的空间，如果所占空间大于hbase.client.write.buffer设置的值，那么就马上调用backgroundFlushCommits(false)方法，否则啥也不做，如果出错就马上调用一次backgroundFlushCommits(true)，所以我们很有必要继续跟踪BufferedMutatorImpl.backgroundFlushCommits(boolean synchronous)代码：
private void backgroundFlushCommits(boolean synchronous) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
LinkedList<Mutation> buffer = new LinkedList<>();
// Keep track of the size so that this thread doesn't spin forever
long dequeuedSize = 0;
try {
//分析所有提交的List<Put>,Put是Mutation的实现
Mutation m;
//如果(hbase.client.write.buffer <= 0 || 0 < (whbase.client.write.buffer * 2) || synchronous)&& writeAsyncBuffer里仍然有Mutation对象
//那么就不断计算所占空间大小dequeuedSize
//currentWriteBufferSize的大小则递减
while ((writeBufferSize <= 0 || dequeuedSize < (writeBufferSize * 2) || synchronous) && (m = writeAsyncBuffer.poll()) != null) {
buffer.add(m);
long size = m.heapSize();
dequeuedSize += size;
currentWriteBufferSize.addAndGet(-size);
}
//backgroundFlushCommits(false)时候，当List<Put>，这里不会进入
if (!synchronous && dequeuedSize == 0) {
return;
}
//backgroundFlushCommits(false)时候，这里会进入,并且不会等待结果返回
if (!synchronous) {
//不会等待结果返回
ap.submit(tableName, buffer, true, null, false);
if (ap.hasError()) {
LOG.debug(tableName + ": One or more of the operations have failed -"
+ " waiting for all operation in progress to finish (successfully or not)");
}
}
//backgroundFlushCommits(true)时候，这里会进入,并且会等待结果返回
if (synchronous || ap.hasError()) {
while (!buffer.isEmpty()) {
ap.submit(tableName, buffer, true, null, false);
}
//会等待结果返回
RetriesExhaustedWithDetailsException error = ap.waitForAllPreviousOpsAndReset(null);
if (error != null) {
if (listener == null) {
throw error;
} else {
this.listener.onException(error, this);
}
}
}
} finally {
//如果还有数据，那么给到外面最后提交
for (Mutation mut : buffer) {
long size = mut.heapSize();
currentWriteBufferSize.addAndGet(size);
dequeuedSize -= size;
writeAsyncBuffer.add(mut);
}
}
}
这里会调用ap.submit(tableName, buffer, true, null, false)直接提交，并且不会等待返回结果，而ap.submit(tableName, buffer, true, null, false)会调用AsyncProcess.submit(ExecutorService pool, TableName tableName,List< extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback,boolean needResults)，这里源代码如下：
public <CResult> AsyncRequestFuture submit(TableName tableName, List< extends Row> rows,
boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults)
throws InterruptedIOException {
return submit(null, tableName, rows, atLeastOne, callback, needResults);
}
public <CResult> AsyncRequestFuture submit(ExecutorService pool, TableName tableName, List< extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults) throws InterruptedIOException {
//如果提交的记录数为0，就直接返回NO_REQS_RESULT
if (rows.isEmpty()) {
return NO_REQS_RESULT;
}
Map<ServerName, MultiAction<Row>> actionsByServer = new HashMap<ServerName, MultiAction<Row>>();
//依据提交的List<Put>的记录数构建retainedActions
List<Action<Row>> retainedActions = new ArrayList<Action<Row>>(rows.size());
NonceGenerator ng = this.connection.getNonceGenerator();
long nonceGroup = ng.getNonceGroup(); // Currently, nonce group is per entire client.
// Location errors that happen before we decide what requests to take.
List<Exception> locationErrors = null;
List<Integer> locationErrorRows = null;
//只要retainedActions不为空，那么就一直执行
do {
// Wait until there is at least one slot for a new task.
// 默认maxTotalConcurrentTasks=100，即最多100个异步线程用于处理元数据获取任务，如果超过100，就等待
waitForMaximumCurrentTasks(maxTotalConcurrentTasks - 1);
// Remember the previous decisions about regions or region servers we put in the
// final multi.
// 记录本次提交的List<Put>对应的region和regionserver
Map<HRegionInfo, Boolean> regionIncluded = new HashMap<HRegionInfo, Boolean>();
Map<ServerName, Boolean> serverIncluded = new HashMap<ServerName, Boolean>();
int posInList = -1;
Iterator< extends Row> it = rows.iterator();
while (it.hasNext()) {
//这里默认传入一个Put对象，因为Put是Row的继承类
Row r = it.next();
//建立变量loc用来存储Put对象对应的region对应的元数据信息
HRegionLocation loc;
try {
if (r == null) {
throw new IllegalArgumentException("#" + id + ", row cannot be null");
}
// Make sure we get 0-s replica.
//取得Put对象对应的region元数据信息的所有备份信息，第一次调用时候会缓存中是没有元数据信息的，那么就会去链接zookeeper上查找，找到后就加入到缓存，下一次直接从缓存中获取
RegionLocations locs = connection.locateRegion(
tableName, r.getRow(), true, true, RegionReplicaUtil.DEFAULT_REPLICA_ID);
if (locs == null || locs.isEmpty() || locs.getDefaultRegionLocation() == null) {
throw new IOException("#" + id + ", no location found, aborting submit for"
+ " tableName=" + tableName + " rowkey=" + Bytes.toStringBinary(r.getRow()));
}
//取得Put对象对应的region元数据信息的所有备份信息数组中的第一个
loc = locs.getDefaultRegionLocation();
} catch (IOException ex) {
locationErrors = new ArrayList<Exception>();
locationErrorRows = new ArrayList<Integer>();
LOG.error("Failed to get region location ", ex);
// This action failed before creating ars. Retain it, but do not add to submit list.
// We will then add it to ars in an already-failed state.
retainedActions.add(new Action<Row>(r, ++posInList));
locationErrors.add(ex);
locationErrorRows.add(posInList);
it.remove();
break; // Backward compat: we stop considering actions on location error.
}
//这里判断是否可以操作，因为最多也就100个异步线程获取元数据信息，如果都忙就等待
if (canTakeOperation(loc, regionIncluded, serverIncluded)) {
Action<Row> action = new Action<Row>(r, ++posInList);
setNonce(ng, r, action);//
retainedActions.add(action);
// TODO: replica-get is not supported on this path
byte[] regionName = loc.getRegionInfo().getRegionName();
//把同一个区的提交任务进行收集，这里先只获知元数据信息，用于知道数据需要提交到哪个region和regionserver，最后循环外再做提交
addAction(loc.getServerName(), regionName, action, actionsByServer, nonceGroup);
it.remove();
}
}
} while (retainedActions.isEmpty() && atLeastOne && (locationErrors == null));
if (retainedActions.isEmpty()) return NO_REQS_RESULT;
// 这里已经知道数据该提交到哪个region和regionserver，就进行批量提交
return submitMultiActions(tableName, retainedActions, nonceGroup, callback, null, needResults, locationErrors, locationErrorRows, actionsByServer, pool);
}
上面代码会去寻找提交的List<Put>的每个Put对象对应的region是哪个，对应的regionserver是哪个，然后进行批量提交，这里要提到另外一个值hbase.client.max.total.tasks(默认值100，意思为客户端最大处理线程数)，如果去请求Put对象对应的region是哪个和对应的regionserver是哪个的操作大于100，那么就要等待，我们回到最初的客户端批量提交代码：
public void put(final List<Put> puts) throws IOException {
//根据设置的缓存大小，达到缓存相关值就进行批量提交
getBufferedMutator().mutate(puts);
//不管有无数据未提交，默认autoFlush=true，那么就最后提交一次
if (autoFlush) {
flushCommits();
}
}
上面的分析可知，如果客户端提交的List<Put>所占空间满足不同条件会进行不同处理，总结如下：
- List<Put>所占空间<hbase.client.write.buffer:getBufferedMutator().mutate(puts)会直接退出，直接执行flushCommits()
- hbase.client.write.buffer<List<Put>所占空间<2*hbase.client.write.buffer:getBufferedMutator().mutate(puts)里面会执行backgroundFlushCommits(false)，处理完后执行flushCommits()
- 2*hbase.client.write.buffer<List<Put>所占空间:getBufferedMutator().mutate(puts)里面会执行backgroundFlushCommits(false),多余的未提交数据会保留，然后执行flushCommits()
紧接着，如果HTable的属性autoFlush（默认为true），那么不管剩下的数据多少，也会进行最后一次提交数据到hbase服务端，这时候flushCommits()里调用的是getBufferedMutator().flush()，而getBufferedMutator().flush()调用的是BufferedMutatorImpl.backgroundFlushCommits(true)，最后调用上面的ap.submit(tableName, buffer, true, null, false)并且会调用ap.waitForAllPreviousOpsAndReset(null)等待返回结果，至此hbase客户端批量提交的源代码分析完毕。
2.5.HConnectionImplementation.locateRegionInMeta
上面的代码HTable.put(final List<Put> puts)分析中我们需要关注另一个重要的信息，就是org.apache.hadoop.hbase.client.AsyncProcess的方法public <CResult> AsyncRequestFuture submit(TableName tableName, List< extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults)，在这个方法里有这么一段代码：
// 获取我们的数据表的region信息
RegionLocations locs = connection.locateRegion(tableName,r.getRow(), true, true, RegionReplicaUtil.DEFAULT_REPLICA_ID);
实质是调用了org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation的方法public RegionLocations locateRegion(final TableName tableName,final byte [] row, boolean useCache, boolean retry, int replicaId)，这个方法加载了我们的hbase数据表的region信息，代码解释如下：
public RegionLocations locateRegion(final TableName tableName, final byte [] row, boolean useCache, boolean retry, int replicaId) throws IOException {
//如果当前连接已经关闭，抛出异常
if (this.closed) throw new IOException(toString() + " closed");
//如果客户端传入hbase数据表为空，抛出异常
if (tableName== null || tableName.getName().length == 0) {
throw new IllegalArgumentException("table name cannot be null or zero length");
}
//TableName.META_TABLE_NAME=hbase:meta(冒号前hbase为包名，meta为表名)
//我们传入的是我们自己的hbase数据表名，而不是hbase:meta,所以这里不会进入
if (tableName.equals(TableName.META_TABLE_NAME)) {
return locateMeta(tableName, useCache, replicaId);
} else {
// 这里的代码会进入
// 这里会去hbase的元数据信息表hbase:meta里去按照我们所给的数据表名和rowkey寻找我们的hbase数据表的region信息
return locateRegionInMeta(tableName, row, useCache, retry, replicaId);
}
}
我们继续关注locateRegionInMeta(tableName, row, useCache, retry, replicaId)，代码注释如下：
/*
* 这里会去hbase的元数据信息表hbase:meta里去按照我们所给的数据表名和rowkey寻找我们的hbase数据表的region信息
*/
private RegionLocations locateRegionInMeta(TableName tableName, byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
// 这里传入的useCache=true，所以会进入
if (useCache) {
//虽然进入了，但是第一次从缓存中找不到我们的数据表的相关信息
RegionLocations locations = getCachedLocation(tableName, row);
if (locations != null && locations.getRegionLocation(replicaId) != null) {
return locations;
}
}
//这里去元数据表hbase:meta中找数据，所以需要构造rowkey
// rowkey=tableName+我们传入的rowkey+"99999999999999"+前面字符的md5HashBytes
byte[] metaKey = HRegionInfo.createRegionName(tableName, row, HConstants.NINES, false);
//这里构造元数据表hbase:meta的查询scan
Scan s = new Scan();
s.setReversed(true);
s.setStartRow(metaKey);
s.setSmall(true);
s.setCaching(1);
if (this.useMetaReplicas) {
s.setConsistency(Consistency.TIMELINE);
}
//默认numTries=31次，无法从元数据表hbase:meta获取信息，那么就一直尝试31次
int localNumRetries = (retry numTries : 1);
for (int tries = 0; true; tries++) {
if (tries >= localNumRetries) {
throw new NoServerForRegionException("Unable to find region for " + Bytes.toStringBinary(row) + " in " + tableName + " after " + localNumRetries + " tries.");
}
if (useCache) {//这里虽然进入了，因为useCache=true,但是我们第一次还是无法从缓存拿到数据
RegionLocations locations = getCachedLocation(tableName, row);
if (locations != null && locations.getRegionLocation(replicaId) != null) {
return locations;
}
} else {
// If we are not supposed to be using the cache, delete any existing cached location
// so it won't interfere.
metaCache.clearCache(tableName, row);
}
// 因为缓存拿不到，那么就从元数据表hbase:meta获取region信息
try {
Result regionInfoRow = null;
ReversedClientScanner rcs = null;
try {
//这里很重要，告诉刚才构造的scan用于表TableName.META_TABLE_NAME，而TableName.META_TABLE_NAME=hbase:meta
rcs = new ClientSmallReversedScanner(conf, s, TableName.META_TABLE_NAME, this, rpcCallerFactory, rpcControllerFactory, getMetaLookupPool(), 0);
//好了，这里拿到了我们的数据表的regionInfoRow信息，regionInfoRow是元数据表hbase:meta中的一行数据
regionInfoRow = rcs.next();
} finally {
if (rcs != null) {
rcs.close();
}
}
if (regionInfoRow == null) {
throw new TableNotFoundException(tableName);
}
// 转换数据表的regionInfoRow信息为我们需要的HRegionLocation
RegionLocations locations = MetaTableAccessor.getRegionLocations(regionInfoRow);
if (locations == null || locations.getRegionLocation(replicaId) == null) {
throw new IOException("HRegionInfo was null in " + tableName + ", row=" + regionInfoRow);
}
//我们拿到了我们的hbase数据表的HRegionLocation，但是此时再做个检查，避免此时hbase宕机了或者已经split了或者拿错了
HRegionInfo regionInfo = locations.getRegionLocation(replicaId).getRegionInfo();
if (regionInfo == null) {
throw new IOException("HRegionInfo was null or empty in " + TableName.META_TABLE_NAME + ", row=" + regionInfoRow);
}
if (!regionInfo.getTable().equals(tableName)) {
throw new TableNotFoundException( "Table '" + tableName + "' was not found, got: " + regionInfo.getTable() + ".");
}
if (regionInfo.isSplit()) {
throw new RegionOfflineException("the only available region for" + " the required row is a split parent," + " the daughters should be online soon: " + regionInfo.getRegionNameAsString());
}
if (regionInfo.isOffline()) {
throw new RegionOfflineException("the region is offline, could" + " be caused by a disable table call: " + regionInfo.getRegionNameAsString());
}
ServerName serverName = locations.getRegionLocation(replicaId).getServerName();
if (serverName == null) {
throw new NoServerForRegionException("No server address listed " + "in " + TableName.META_TABLE_NAME + " for region " + regionInfo.getRegionNameAsString() + " containing row " + Bytes.toStringBinary(row));
}
if (isDeadServer(serverName)){
throw new RegionServerStoppedException("hbase:meta says the region "+ regionInfo.getRegionNameAsString()+" is managed by the server " + serverName + ", but it is dead.");
}
// 好了检查无误了，那么为了让下一次不要这么麻烦，先缓存起来，这样拿的也快
cacheLocation(tableName, locations);
// 好了，该返回region信息了
return locations;
} catch (TableNotFoundException e) {
// if we got this error, probably means the table just plain doesn't
// exist. rethrow the error immediately. this should always be coming
// from the HTable constructor.
throw e;
} catch (IOException e) {
ExceptionUtil.rethrowIfInterrupt(e);
if (e instanceof RemoteException) {
e = ((RemoteException)e).unwrapRemoteException();
}
if (tries < localNumRetries - 1) {
if (LOG.isDebugEnabled()) {
LOG.debug("locateRegionInMeta parentTable=" + TableName.META_TABLE_NAME + ", metaLocation=" + ", attempt=" + tries + " of " + localNumRetries + " failed; retrying after sleep of " + ConnectionUtils.getPauseTime(this.pause, tries) + " because: " + e.getMessage());
}
} else {
throw e;
}
// Only relocate the parent region if necessary
if(!(e instanceof RegionOfflineException || e instanceof NoServerForRegionException)) {
relocateRegion(TableName.META_TABLE_NAME, metaKey, replicaId);
}
}
//没找到，那么沉睡一段时间然后重试次数未到31次，那么继续循环找吧，直到找到，如果次数大于31，那么只有抛出异常
try{
Thread.sleep(ConnectionUtils.getPauseTime(this.pause, tries));
} catch (InterruptedException e) {
throw new InterruptedIOException("Giving up trying to location region in " + "meta: thread is interrupted.");
}
}
}
上述代码我们可以得知在首次org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation是如何加载我们需要的hbase数据表的信息的，我们看到hbase有个元数据表hbase:meta，这里hbase是namespace而meta是表名，我们自己创建的数据表的元数据信息都存储在这个元数据表hbase:meta中，第一次的时候会去元数据表hbase：meta中查找，找到后就加入缓存，第二次的时候直接从缓存获取我们的数据表的region信息
3.从分析源码中学到的对于hbase客户端的优化知识
- hbase客户端里传入hbase.client.write.buffer(默认2MB)，加到客户端提交的缓存大小；
- hbase客户端提交采用批量提交，批量提交的List<Put>的size计算公式=hbase.client.write.buffer*2/Put大小，Put大小可通过put.heapSize()获取，以hbase.client.write.buffer=2097152，put.heapSize()=1320举例，最佳的批量提交记录大小=2*2097152/1320=3177;
- hbase客户端尽量采用多线程并发写
- hbase客户端所在机器性能要好，不然速度上不去
- 能接受关闭WAL的话尽量关闭，速度也会相应提升
4.hbase性能调研写入速度测试记录
