科学使用HBase Connection

2019-09-02 16:20:40 浏览数 (1)

为什么要写这篇小文章?因为偶然在技术交流群里看到了如下的问题。

这个问题的答案简单而不简单:HBase客户端是不需要维护连接池的,或者说,Connection对象已经帮我们做好了。但是,对Connection使用不当是HBase新手(包括很久很久之前的我自己)最容易犯的错误之一,常见错误用法有:

  • 每个线程开一个连接,线程结束时关闭;
  • 每次读写HBase时开一个连接,读写完毕后关闭;
  • 自行实现Connection对象的池化,每次使用时取出一个。

之前已经多次提到过,创建HBase连接是非常“贵”(expensive)的操作,并且创建过多的Connection会导致HBase拒绝连接。因此,最科学的方式就是在整个应用(进程)的范围内只维护一个共用的Connection,比如以单例的形式。在应用退出时,再关闭连接。

下面不妨来深入地看看Connection是怎么维护连接的,毕竟它与我们平时了解到的JDBC连接等有很大的不同。来看看org.apache.hadoop.hbase.client.Connection这个接口的源码。

代码语言:javascript复制
/**
 * 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连接,该方法有多种重载,最底层的重载如下。

代码语言:javascript复制
  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()方法创建了它。

代码语言:javascript复制
private RpcClient rpcClient;
this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);

这个方法也是通过反射来创建RPC客户端的(HBase里到处都是反射),实例化的类为BlockingRpcClient,它是AbstractRpcClient抽象类的子类。AbstractRpcClient中使用了一个名为PoolMap的结构来维护ConnectionId与连接池之间的映射关系,在构造方法中初始化。

代码语言:javascript复制
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与服务名称三者的包装类。

代码语言:javascript复制
  public ConnectionId(User ticket, String serviceName, InetSocketAddress address) {
    this.address = address;
    this.ticket = ticket;
    this.serviceName = serviceName;
  }

接下来注意到AbstractRpcClient.getConnection()方法。

代码语言:javascript复制
  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确实是重量级的玩意儿,有一个就够了。

0 人点赞