欢迎扫码关注我的VX公众号,回复【JAVAPDF】可以获得一份200页秋招面试题~
为什么要写这篇小文章?因为偶然在技术交流群里看到了如下的问题。
这个问题的答案简单而不简单: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, InetSocket address 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确实是重量级的玩意儿,有一个就够了。
— THE END —