hbase-0.92.1-cdh4.1.3的HTablePool实现:

?

1. PoolType:
Reusable: （默认）一个实例池，多线程复用，内部用ConcurrentLinkedQueue装多个实例HTable;
ThreadLocal: 每个线程只有一个实例,线程与线程之间互不影响, ThreadLocal;
特点是随着线程的增多,Pool中的HTable增多,但互不影响;
RoundRobin: Pool中的HTable用CopyOnWriteArrayList装;

2. 初始化:
HTablePool pool = new HTablePool(conf, 5);
// 默认 PoolType.Reusable
pool = new HTablePool(conf, maxSize, tableFactory, PoolMap.PoolType.ThreadLocal);
// PoolMap.PoolType.ThreadLocal
pool = new HTablePool(conf, maxSize, tableFactory, PoolMap.PoolType.RoundRobin);
// PoolMap.PoolType.RoundRobin
实例化PoolMap
实例化HTablePool，此时还没有任何HTable实例?，tables为空;

3. 取得HTableInterface对象:
pool.getTable(TEST_TABLE_NAME);?
查看tables是否含有table,如果没有，创建一个HTable实例
返回HTable实例封装成PooledHTable实例返回
PooledHTable实例用完.close();后放置到PoolMap;

4. HTablePool可以容纳任何的Table的HTableInterface实例.
HTable实例会共享同一个zookeeper连接
HTable实例，如果同在一个RegionServer会共享同一个连接HBaseClient$Connection
HTablePool有最大尺寸，但并没有限制HTable实例不得大于这个尺寸，一旦超过这个尺寸就会实例化，但归还到实例池的时候，如果池满了会弃用;
HTable实例线程不安全;

?

注意点:
1. 在多线程使用HTablePool拿到同一个表的HTable时，如果线程个数大于maxsize会导致写入始终是autoflush！

public HTableInterface getTable(String tableName) {
   // call the old getTable implementation renamed to findOrCreateTable
   HTableInterface table = findOrCreateTable(tableName);
   // return a proxy table so when user closes the proxy, the actual table
   // will be returned to the pool
   return new PooledHTable(table);
}
public void close() throws IOException {
   returnTable(table);
}
private void returnTable(HTableInterface table) throws IOException {
   // this is the old putTable method renamed and made private
   String tableName = Bytes.toString(table.getTableName());
   if (tables.size(tableName) >= maxSize) {
     // release table instance since we're not reusing it
     this.tables.remove(tableName, table);
     this.tableFactory.releaseHTableInterface(table);
     return;
   }
   tables.put(tableName, table);
}

?

如果tables.size大于maxsize，此时会去掉一个保存的HTable对象，而releaseHTableInterface实际调用的就是HTable的close方法，close方法又会强制flushHTable的buffer，因此，如果我们想不使用autoflush提升写入速度失效。

2. 改写HTable,的flushCommit(固定频率+内存占用>1M)

@Override
public void put(final List<Put> puts) throws IOException {
    super.put(puts);
    needFlush();
}
private void needFlush() throws IOException {
    long currentTime = System.currentTimeMillis();
    if ((currentTime - lastFlushTime.longValue()) > flushInterval) {
        super.flushCommits();
        lastFlushTime.set(currentTime);
    }
}

?

初始化使用代码样例

?

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.BeforeClass;
import org.junit.Test;

public class HTablePoolTest2 {

	protected static String TEST_TABLE_NAME = "testtable";

	protected static String ROW1_STR = "row1";
	protected static String COLFAM1_STR = "colfam1";
	protected static String QUAL1_STR = "qual1";

	private final static byte[] ROW