科学使用HBase Connection
为什么要写这篇小文章？因为偶然在技术交流群里看到了如下的问题。
这个问题的答案简单而不简单：HBase客户端是不需要维护连接池的，或者说，Connection对象已经帮我们做好了。但是，对Connection使用不当是HBase新手（包括很久很久之前的我自己）最容易犯的错误之一，常见错误用法有：
- 每个线程开一个连接，线程结束时关闭；
- 每次读写HBase时开一个连接，读写完毕后关闭；
- 自行实现Connection对象的池化，每次使用时取出一个。
之前已经多次提到过，创建HBase连接是非常“贵”（expensive）的操作，并且创建过多的Connection会导致HBase拒绝连接。因此，最科学的方式就是在整个应用（进程）的范围内只维护一个共用的Connection，比如以单例的形式。在应用退出时，再关闭连接。
下面不妨来深入地看看Connection是怎么维护连接的，毕竟它与我们平时了解到的JDBC连接等有很大的不同。来看看org.apache.hadoop.hbase.client.Connection这个接口的源码。
/**
* A cluster connection encapsulating lower level individual connections to actual servers and
* a connection to zookeeper. Connections are instantiated through the {@link ConnectionFactory}
* class. The lifecycle of the connection is managed by the caller, who has to {@link #close()}
* the connection to release the resources.
*
* <p> The connection object contains logic to find the master, locate regions out on the cluster,
* keeps a cache of locations and then knows how to re-calibrate after they move. The individual
* connections to servers, meta cache, zookeeper connection, etc are all shared by the
* {@link Table} and {@link Admin} instances obtained from this connection.
*
* <p> Connection creation is a heavy-weight operation. Connection implementations are thread-safe,
* so that the client can create a connection once, and share it with different threads.
* {@link Table} and {@link Admin} instances, on the other hand, are light-weight and are not
* thread-safe. Typically, a single connection per client application is instantiated and every
* thread will obtain its own Table instance. Caching or pooling of {@link Table} and {@link Admin}
* is not recommended.
*
* <p>This class replaces {@link HConnection}, which is now deprecated.
* @see ConnectionFactory
* @since 0.99.0
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface Connection extends Abortable, Closeable {
Configuration getConfiguration();
Table getTable(TableName tableName) throws IOException;
Table getTable(TableName tableName, ExecutorService pool) throws IOException;
public BufferedMutator getBufferedMutator(TableName tableName) throws IOException;
public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws IOException;
public RegionLocator getRegionLocator(TableName tableName) throws IOException;
Admin getAdmin() throws IOException;
@Override
public void close() throws IOException;
boolean isClosed();
}这里特意把它的JavaDoc留下来了，因为这段文档的信息量比较大。我们可以得出如下结论：
- Connection对象需要知道如何找到HMaster、如何在RegionServer上定位Region，以及感知Region的变动。所以，Connection需要同时与HMaster、RegionServer和ZK建立连接。
- 创建Connection是重量级的，并且它是线程安全的。
- 由Connection取得的Table和Admin对象是轻量级的，并且不是线程安全的，所以它们应该即用即弃。
我们几乎都是通过调用ConnectionFactory.createConnection()方法来创建HBase连接，该方法有多种重载，最底层的重载如下。
static Connection createConnection(final Configuration conf, final boolean managed,
final ExecutorService pool, final User user)
throws IOException {
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 {
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对象。后面的代码就越发地复杂了（这就是为什么写源码阅读专题选了Spark而没选HBase），为了避免篇幅过长，只列出最关键的逻辑。
HBase中的RPC客户端由RpcClient接口的子类来实现，在HConnectionImplementation的构造方法中，调用RpcClientFactory.createClient()方法创建了它。
private RpcClient rpcClient;
this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);
这个方法也是通过反射来创建RPC客户端的（HBase里到处都是反射），实例化的类为BlockingRpcClient，它是AbstractRpcClient抽象类的子类。AbstractRpcClient中使用了一个名为PoolMap的结构来维护ConnectionId与连接池之间的映射关系，在构造方法中初始化。
protected final PoolMap<ConnectionId, T> connections;
this.connections = new PoolMap<>(getPoolType(conf), getPoolSize(conf));
PoolMap中的连接池的类型是Pool<T>，T则是连接对象的类型。Pool的具体类型由参数hbase.client.ipc.pool.type来确定，可选RoundRobinPool、ThreadLocalPool与ReusablePool三种，默认为第一种。连接池大小则由参数hbase.client.ipc.pool.size来确定，默认值为1。
ConnectionId也并不是一个简单的ID，而是服务器地址、用户ticket与服务名称三者的包装类。
public ConnectionId(User ticket, String serviceName, InetSocketAddress address) {
this.address = address;
this.ticket = ticket;
this.serviceName = serviceName;
}接下来注意到AbstractRpcClient.getConnection()方法。
private T getConnection(ConnectionId remoteId) throws IOException {
if (failedServers.isFailedServer(remoteId.getAddress())) {
if (LOG.isDebugEnabled()) {
LOG.debug("Not trying to connect to " + remoteId.address
+ " this server is in the failed servers list");
}
throw new FailedServerException(
"This server is in the failed servers list: " + remoteId.address);
}
T conn;
synchronized (connections) {
if (!running) {
throw new StoppedRpcClientException();
}
conn = connections.get(remoteId);
if (conn == null) {
conn = createConnection(remoteId);
connections.put(remoteId, conn);
}
conn.setLastTouched(EnvironmentEdgeManager.currentTime());
}
return conn;
}该方法检查connections映射中是否已有对应ID的连接池，如果有，就直接返回；没有的话，就调用子类实现的createConnection()方法创建一个（类型为BlockingRpcConnection），将其放入connections映射并返回。由此可见，Connection对象确实替我们维护了所有的连接。
经由上面的简单分析，可以总结出如下的图。由此可见，Connection确实是重量级的玩意儿，有一个就够了。
