转载请标注原链接：http://www.cnblogs.com/xczyd/p/5577124.html
客户在使用HBase的时候，经常会抱怨说写入太慢，并发上不去等等。从前我遇到这种情况，一般都二话不说，直接去看HBase集群的负载，看看有什么性能瓶颈等等。
某老司机说，且慢，先看看用户怎么写的客户端访问HBase集群的代码。
于是花了一些时间去看。
不看不知道，一看就吓尿。客户（也包括我们自己的实施）写出来的客户端，很多时候存在很多低级错误，比如：
（1）滥用sychronize；
（2）创建了连接不释放；
（3）明明只需要调用一次的API，却进行了多次调用，要是碰巧遇到比较花时间的API，那性能就可想而知了；
（4）其他各种幺蛾子...
为此，本篇仅从HBase的Java API入手，通过源码分析和简单的实验，找到最合适Java API调用方法（主要服务于高并发场景）。
如果对HBase的Java API不熟悉的话，可以先去官网看一下文档。
下面开始正文：
使用Java API与HBase集群交互时，需要先创建一个HTable的实例，再使用该实例提供的方法来进行插入/删除/查询等操作。
要创建HTable对象，要先创建一个包含了HBase集群信息的配置实例Configuration conf，其一般创建方法如下：
Configuration conf = HBaseConfiguration.create();
//设置HBase集群的IP和端口
conf.set("hbase.zookeeper.quorum", "XX.XXX.X.XX");
conf.set("hbase.zookeeper.property.clientPort", "2181");
在拥有了conf之后，可以通过HTable提供的如下两种构造方法来创建HTable实例：
方法一：直接利用conf来创建HTable实例
对应的构造函数如下：
public HTable(Configuration conf, final TableName tableName)
throws IOException {
this.tableName = tableName;
this.cleanupPoolOnClose = this.cleanupConnectionOnClose = true;
if (conf == null) {
this.connection = null;
return;
}
this.connection = HConnectionManager.getConnection(conf);
this.configuration = conf;
this.pool = getDefaultExecutor(conf);
this.finishSetup();
}
注意红色部分的代码。在这种构造方法中，会调用HConnectionManager的getConnection函数，这个函数以conf作为输入参数，来获取了一个HConnection的实例connection。熟悉odbc，jdbc的话，会知道使用Java API进行数据库操作的时候，都会创建一个类似的connection/connection pool来维护一些数据库与客户端之间相互的连接。对于Hbase来说，承担类似角色的就是HConnection。不过与oracle不同的一点是，HConnection实际上去连接的并不是HBase集群本身，而是维护其关键数据信息的Zookeeper（简称ZK）集群。有关ZK的内容在这里不做展开，不熟悉的话可以单纯地理解为一个独立的元信息管理角色。回过来看getConnection函数，其具体实现如下：
public static HConnection getConnection(final Configuration conf)
throws IOException {
HConnectionKey connectionKey = new HConnectionKey(conf);
synchronized (CONNECTION_INSTANCES) {
HConnectionImplementation connection = CONNECTION_INSTANCES.get(connectionKey);
if (connection == null) {
connection = (HConnectionImplementation)createConnection(conf, true);
CONNECTION_INSTANCES.put(connectionKey, connection);
} else if (connection.isClosed()) {
HConnectionManager.deleteConnection(connectionKey, true);
connection = (HConnectionImplementation)createConnection(conf, true);
CONNECTION_INSTANCES.put(connectionKey, connection);
}
connection.incCount();
return connection;
}
}
其中，CONNECTION_INSTANCES的类型是LinkedHashMap<HConnectionKey，HConnectionImplementation>。所谓HConnectionImplementation其实就是HConnection的具体实现。继续注意红色部分的三行代码。第一行，通过conf创建了一个HConnectionKey的实例connectionKey；第二行，去CONNECTION_INSTANCES中查找是否存在与connectionKey对应的一个HConnection的实例；第三行，如果不存在，那么调用createConnection来创建一个HConnection的实例，否则直接返回刚才从Map中查找得到的HConnection对象
不嫌麻烦，再看一下HConnectionKey的构造函数和重写的hashCode函数，代码分别如下：
HConnectionKey(Configuration conf) {
Map<String, String> m = new HashMap<String, String>();
if (conf != null) {
for (String property : CONNECTION_PROPERTIES) {
String value = conf.get(property);
if (value != null) {
m.put(property, value);
}
}
}
this.properties = Collections.unmodifiableMap(m);
try {
UserProvider provider = UserProvider.instantiate(conf);
User currentUser = provider.getCurrent();
if (currentUser != null) {
username = currentUser.getName();
}
} catch (IOException ioe) {
HConnectionManager.LOG.warn("Error obtaining current user, skipping username in HConnectionKey", ioe);
}
}
public int hashCode() {
final int prime = 31;
int result = 1;
if (username != null) {
result = username.hashCode();
}
for (String property : CONNECTION_PROPERTIES) {
String value = properties.get(property);
if (value != null) {
result = prime * result + value.hashCode();
}
}
return result;
}
可以看到，hashCode函数被重写以后，其返回值实际上是username的hashCode函数的返回值，而username来自于currentuser，currentuser又来自于provider，provider是由conf创建的。可以看出，只要有相同的conf，就能创建出相同的username，也就能保证HConnectionKey的hashCode函数被重写以后，能够在username相同时返回相同的值。而CONNECTION_INSTANCES是一个LinkedHashMap，其get函数会调用HConnectionKey的hashCode函数来判断该对象是否已经存在。因此，getConnection函数的本质就是根据conf信息返回connection对象，对每一个内容相同的conf，只会返回一个connection
方法二：调用createConnection方法来显式地创建Hconnection的实例，再将其作为输入参数来创建HTable实例
createConnection方法和Htable对应的构造函数分别如下：
public static HConnection createConnection(Configuration conf) throws IOException {
UserProvider provider = UserProvider.instantiate(conf);
return createConnection(conf, false, null, provider.getCurrent());
}
static HConnection createConnection(final Configuration conf, final boolean managed,final ExecutorService pool, final User user)
throws IOException { String className = conf.get("hbase.client.connection.impl",HConnectionManager.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.
Constructor<?> constructor =
clazz.getDeclaredConstructor(Configuration.class,
boolean.class, ExecutorService.class, User.class);
constructor.setAccessible(true);
return (HConnection) constructor.newInstance(conf, managed, pool, user);
} catch (Exception e) {
throw new IOException(e);
}
}
public HTable(TableName tableName, HConnection connection) throws IOException {
this.tableName = tableName;
this.cleanupPoolOnClose = true;
this.cleanupConnectionOnClose = false;
this.connection = connection;
this.configuration = connection.getConfiguration();
this.pool = getDefaultExecutor(this.configuration);
this.finishSetup();
}
可以看出，这种构造HTable的方法会通过反射来创建一个新的HConnection实例，而不像方法一中那样共享一个HConnection实例。
值得一提的是，通过此种方法创建出来的HConnection，是需要在不再使用的时候显式调用close方法去释放掉的，否则容易造成端口占用等问题。
那么，上述两种方法，在执行插入/删除/查找的时候，性能如何呢？不妨先从代码角度分析一下。为了简便，先分析HTable在执行put(插入)操作时具体做的事情。
HTable的put函数如下：
public void put(final Put put) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
doPut(put);
if (autoFlush) {
flushCommits();
}
}
private void doPut(Put put) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
if (ap.hasError()){
writeAsyncBuffer.add(put);
backgroundFlushCommits(true);
}
validatePut(put);
currentWriteBufferSize += put.heapSize();
writeAsyncBuffer.add(put);
while (currentWriteBufferSize > writeBufferSize) {
backgroundFlushCommits(false);
}
}
private void backgroundFlushCommits(boolean synchronous) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
try {
do {
ap.submit(writeAsyncBuffer, true);
} while (synchronous && !writeAsyncBuffer.isEmpty());
if (synchronous) {
ap.waitUntilDone();
}
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)");
while (!writeAsyncBuffer.isEmpty()) {
ap.submit(writeAsyncBuffer, true);
}
ap.waitUntilDone();
if (!clearBufferOnFail) {
// if clearBufferOnFailed is not set, we're supposed to keep the failed operation in the
// write buffer. This is a questionable feature kept here for backward compatibility
writeAsyncBuffer.addAll(ap.getFailedOperations());
}
RetriesExhaustedWithDetailsException e = ap.getErrors();
ap.clearErrors();
throw e;
}
} finally {
currentWriteBufferSize = 0;
for (Row mut : writeAsyncBuffer) {
if (mut instanceof Mutation) {
currentWriteBufferSize += ((Mutation) mut).heapSize();
}
}
}
}
如红色部分所表示，调用顺序是put->doPut->backgroundFlushCommits->ap.submit，其中ap是类AsyncProcess的对象。因此追踪到AsyncProcess类，其代码如下：
public void submit(List<? extends Row> rows, boolean atLeastOne) throws InterruptedIOException {
submitLowPriority(rows, atLeastOne, false);
}
public void submitLowPriority(List<? extends Row> rows, boolean atLeastOne, boolean isLowPripority) throws InterruptedIOException {
if (rows.isEmpty()) {
return;
}
// This looks like we are keying by region but HRegionLocation has a comparator that compares
// on the server portion only (hostname + port) so this Map collects regions by server.
Map<HRegionLocation, MultiAction<Row>> actionsByServer = new HashMap<HRegionLocation, MultiAction<Row>>();
List<Action<Row>> retainedActions = new ArrayList<Action<Row>>(rows.size());
long currentTaskCnt = tasksDone.get();
boolean alreadyLooped = false;
NonceGenerator ng = this.hConnection.getNonceGenerator();
do {
if (alreadyLooped){
// if, for whatever reason, we looped, we want to be sure that something has changed.
waitForNextTaskDone(currentTaskCnt);
currentTaskCnt = tasksDone.get();
} else {
alreadyLooped = true;
}
// Wait until there is at least one slot for a new task.
waitForMaximumCurrentTasks(maxTotalConcurrentTasks - 1);
// Remember the previous decisions about regions or region servers we put in the
// final multi.
Map<Long, Boolean> regionIncluded = new HashMap<Long, Boolean>();
Map<ServerName, Boolean> serverIncluded = new HashMap<ServerName, Boolean>();
int posInList = -1;
Iterator<? extends Row> it = rows.iterator();
while (it.hasNext()) {
Row r = it.next();
HRegionLocation loc = findDestLocation(r, posInList);
if (loc == null) { // loc is null if there is an error such as meta not available.
it.remove();
} else if (canTakeOperation(loc, regionIncluded, serverIncluded)) {
Action<Row> action = new Action<Row>(r, ++posInList);
setNonce(ng, r, action);
retainedActions.add(action);
addAction(loc, action, actionsByServer, ng);
it.remove();
}
}
} while (retainedActions.isEmpty() && atLeastOne && !hasError());
HConnectionManager.ServerErrorTracker errorsByServer = createServerErrorTracker();
sendMultiAction(retainedActions, actionsByServer, 1, errorsByServer, isLowPripority);
}
private HRegionLocation findDestLocation(Row row, int posInList) {
if (row == null) throw new IllegalArgumentException("#" + id + ", row cannot be null");
HRegionLocation loc = null;
IOException locationException = null;
try {
loc = hConnection.locateRegion(this.tableName, row.getRow());
if (loc == null) {
locationException = new IOException("#" + id + ", no location found, aborting submit for" +
" tableName=" + tableName +
" rowkey=" + Arrays.toString(row.getRow()));
}
} catch (IOException e) {
locationException = e;
}
if (locationException != null) {
// There are multiple retries in locateRegion already. No need to add new.
// We can't continue with this row, hence it's the last retry.
manageError(posInList, row, false, locationException, null);
return null;
}
return loc;
}
这里代码的主要实现机制是异步调用，也就是说，并非每一次put操作都是直接往HBase里面写数据的，而是等到缓存区域内的数据多到一定程度（默认设置是２Ｍ），再进行一次写操作。当然这次操作在Server端应当还是要排队执行的，具体执行机制这里不作展开。可以确定的是，HConnection在插入/查询/删除的Java API中，只是起到一个定位RegionServer的作用，在定位到RegionServer之后，操作都是由client端通过rpc调用完成的，与客户端创建的connection的数目无关。另外，locateRegion其实只有在没有命中缓存的时候才会进行rpc通信，其他时候都是直接从缓存中获取RegionServer信息，详情可以查看locateRegion的源码，这里也不再展开。
代码分析告一段落，通过分析可以确定，createConnection的方法创建出大量的HConnection并不会对写入性能有任何帮助。相反，由于白白浪费了资源，还会比getConnection更慢。但是慢多少，无法仅凭代码作出判断。
不妨简单做一个实验来验证上述论断：
服务器环境：四台linux服务器组成的HBase集群， 内存64G，ping一次平均约5ms（严谨一点的话应该再提供一下cpu核数、频率，以及磁盘转速等信息）
客户端环境：在Mac上装的ubuntu虚拟机，分配内存10G，CPU、网络和磁盘读写速度都要比物理机慢不少，但是不影响结论
实验代码如下：
public class HbaseConectionTest {
public static void main(String[] args) throws Exception{
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "XX.XXX.X.XX");
conf.set("hbase.zookeeper.property.clientPort", "2181");
ThreadInfo info = new ThreadInfo();
info.setTableNamePrefix("test");
info.setColNames("col1,col2");
info.setTableCount(1);
info.setConnStrategy("CREATEWITHCONF");//CREATEWITHCONF,CREATEWITHCONN
info.setWriteStrategy("SEPERATE");//OVERLAP,SEPERATE
info.setLifeCycle(60000L);
int threadCount = 100;
for(int i=0;i<threadCount;i++){
//createTable(tableNamePrefix+i,colNames,conf);
}
//
for(int i=0;i<threadCount;i++){
new Thread(new WriteThread(conf,info,i)).start();
}
//HBaseAdmin admin = new HBaseAdmin(conf);
//System.out.println(admin.tableExists("test"));
}
public static void createTable(String tableName,String[] colNames,Configuration conf) {
System.out.println("start create table "+tableName);
try {
HBaseAdmin hBaseAdmin = new HBaseAdmin(conf);
if (hBaseAdmin.tableExists(tableName)) {
System.out.println(tableName + " is exist");
//hBaseAdmin.disableTable(tableName);
//hBaseAdmin.deleteTable(tableName);
return;
}
HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
for(int i=0;i<colNames.length;i++) {
tableDescriptor.addFamily(new HColumnDescriptor(colNames[i]));
}
hBaseAdmin.createTable(tableDescriptor);
} catch (Exception ex) {
ex.printStackTrace();
}
System.out.println("end create table "+tableName);
}
}
//Thread执行操作的配置信息
class ThreadInfo {
private int tableCount;
String tableNamePrefix;
String[] colNames;
//CREATEBYCONF or CREATEBYCONN
String connStrategy;
//overlap or seperate
String writeStrategy;
long lifeCycle;
public ThreadInfo(){
}
public int getTableCount() {
return tableCount;
}
public void setTableCount(int tableCount) {
this.tableCount = tableCount;
}
public String getTableNamePrefix() {
return tableNamePrefix;
}
public void setTableNamePrefix(String tableNamePrefix) {
this.tableNamePrefix = tableNamePrefix;
}
public String[] getColNames() {
return colNames;
}
public void setColNames(String[] colNames) {
this.colNames = colNames;
}
public void setColNames(String colNames) {
if(colNames == null){
this.colNames = null;
}
else{
this.colNames = colNames.split(",");
}
}
public String getWriteStrategy() {
return writeStrategy;
}
public void setWriteStrategy(String writeStrategy) {
this.writeStrategy = writeStrategy;
}
public String getConnStrategy() {
return connStrategy;
}
public void setConnStrategy(String connStrategy) {
this.connStrategy = connStrategy;
}
public long getLifeCycle() {
return lifeCycle;
}
public void setLifeCycle(long lifeCycle) {
this.lifeCycle = lifeCycle;
}
}
class WriteThread implements Runnable{
private Configuration conf;
private ThreadInfo info;
private int index;
public WriteThread(Configuration conf,ThreadInfo info,int index){
this.conf = conf;
this.info = info;
this.index = index;
}
@Override
public void run(){
String threadName = Thread.currentThread().getName();
int operationCount = 0;
HTable[] htables = null;
HConnection conn = null;
int tableCount = info.getTableCount();
String tableNamePrefix = info.getTableNamePrefix();
String[] colNames = info.getColNames();
String connStrategy = info.getConnStrategy();
String writeStrategy = info.getWriteStrategy();
long lifeCycle = info.getLifeCycle();
System.out.println(threadName+": started with index "+index);
try{
if (connStrategy.equals("CREATEWITHCONN")) {
conn = HConnectionManager.createConnection(conf);
if (writeStrategy.equals("SEPERATE")) {
htables = new HTable[1];
htables[0] = new HTable(TableName.valueOf(tableNamePrefix+(index%tableCount)), conn);
}
else if(writeStrategy.equals("OVERLAP")) {
htables = new HTable[tableCount];
for (int i = 0; i < tableCount; i++) {
htables[i] = new HTable(TableName.valueOf(tableNamePrefix+i), conn);
}
}
else{
return;
}
}
else if (connStrategy.equals("CREATEWITHCONF")) {
conn = null;
if (writeStrategy.equals("SEPERATE")) {
htables = new HTable[1];
htables[0] = new HTable(conf,TableName.valueOf(tableNamePrefix+(index%tableCount)));
}
else if(writeStrategy.equals("OVERLAP")) {
htables = new HTable[tableCount];
for (int i = 0; i < tableCount; i++) {
htables[i] = new HTable(conf,TableName.valueOf(tableNamePrefix+i));
}
}
else{
return;
}
}
else {
return;
}
long start = System.currentTimeMillis();
long end = System.currentTimeMillis();
Map<HTable,HColumnDescriptor[]> table_columnFamilies = new HashMap<HTable,HColumnDescriptor[]>();
for(int i=0;i<htables.length;i++){
table_columnFamilies.put(htables[i],htables[i].getTableDescriptor().getColumnFamilies());
}
while(end-start<=lifeCycle){
HTable table = htables.length==1?htables[0]:htables[(int)Math.random()*htables.length];
long s1 = System.currentTimeMillis();
double r = Math.random();
HColumnDescriptor[] columnFamilies = table_columnFamilies.get(table);
Put put = generatePut(threadName,columnFamilies,colNames,operationCount);
table.put(put);
if(r>0.999){
System.out.println(System.currentTimeMillis()-s1);
}
operationCount++;
end = System.currentTimeMillis();
}
if(conn != null){
conn.close();
}
}catch(Exception ex){
ex.printStackTrace();
}
System.out.println(threadName+": ended with operation count:"+operationCount);
}
private Put generatePut(String threadName,HColumnDescriptor[] columnFamilies,String[] colNames,int operationCount){
Put put = new Put(Bytes.toBytes(threadName+"_"+operationCount));
for (int i = 0; i < columnFamilies.length; i++) {
String familyName = columnFamilies[i].getNameAsString();
//System.out.println("familyName:"+familyName);
for(int j=0;j<colNames.length;j++){
if(familyName.equals(colNames[j])) { //
String columnName = familyName+(int)(Math.floor(Math.random()*5+10*j));
String val = ""+columnName.hashCode()%100;
put.add(Bytes.toBytes(familyName),Bytes.toBytes(columnName),Bytes.toBytes(val));
}
}
}
//System.out.println(put.toString());
return put;
}
}
简单来说就是先创建一些有两列的HBase表，然后创建一些线程分别采用getConnection策略和createConnection策略来写１分钟的数据。当然写几张表，写多久，写什么，怎么写都可以调整。比如我这里就设计了固定写一张表或者随机写一张表几种逻辑。需要注意一下红色部分的代码，这里预先获得了要写的HBase表的列信息。做这个动作的原因是getTableDescriptor是会产生网络开销的，建议写代码时尽量少调用，以免增加不必要的额外开销（事实上这个额外开销是很巨大的）。
具体实验数据如下表所示，具体值因为网络波动等原因会有所差异。总的来说，在并发较高（线程数大于30）的时候，getConnection方法速度要明显快于createConnection；在并发较低的(线程数小于等于10)的时候，createConnection则稍微占优。另外，使用getConnection的时候，写一张表的速度在高并发场景下要明显快于写多张表，但是在低并发情况下此现象不明显；使用createConnection的时候，无论并发高低，写一张表的速度与写多张表大致相同，甚至还偏慢。
上述现象与代码分析的结果并不完全一致。不一致的地方主要包括如下两点：
（1）为什么线程少的时候，createConnection占优？理论上应该持平才是。这一点无法得到很合理的解释，存疑；
（2）为什么线程很多的时候，createConnection会慢这么多？这里猜测服务端的ZK要维护大量连接会负载过大，即便是多个regionServer在负责具体的写操作，也仍旧会导致性能下降。
这两个疑点还有待进一步论证。尽管如此，还是可以先建议大家在使用Java API与HBase交互时，尤其是处理高并发场景的时候，尽量使用getConnection的办法去创建HTable对象，避免维护不必要的connection导致浪费资源。
| thread_count | table_count | conn_strategy | write_strategy | interval | result |
| 1 | 1 | CONF | OVERLAP | 60s | 10000*1=10000 |
| 5 | 1 | CONF | OVERLAP | 60s | 11000*5=55000 |
| 10 | 1 | CONF | OVERLAP | 60s | 12000*10=120000 |
| 30 | 1 | CONF | OVERLAP | 60s | 8300*30=249000 |
| 60 | 1 | CONF | OVERLAP | 60s | 6000*60=360000 |
| 100 | 1 | CONF | OVERLAP | 60s | 4700*100=470000 |
| 1 | 1 | CONN | OVERLAP | 60s | 12000*1=12000 |
| 5 | 1 | CONN | OVERLAP | 60s | 16000*5=80000 |
| 10 | 1 | CONN | OVERLAP | 60s | 10000*10=100000 |
| 30 | 1 | CONN | OVERLAP | 60s | 2500*30=75000 |
| 60 | 1 | CONN | OVERLAP | 60s | 1200*60=72000 |
| 100 | 1 | CONN | OVERLAP | 60s | 1000*100=100000 |
| 5 | 5 | CONF | SEPERATE | 60s | 10600*5=53000 |
| 10 | 10 | CONF | SEPERATE | 60s | 11900*10=119000 |
| 30 | 30 | CONF | SEPERATE | 60s | 6900*30=207000 |
| 60 | 60 | CONF | SEPERATE | 60s | 3650*60=219000 |
| 100 | 100 | CONF | SEPERATE | 60s | 2500*100=250000 |
| 5 | 5 | CONN | SEPERATE | 60s | 14000*5=70000 |
| 10 | 10 | CONN | SEPERATE | 60s | 10500*10=105000 |
| 30 | 30 | CONN | SEPERATE | 60s | 3250*30=97500 |
| 60 | 60 | CONN | SEPERATE | 60s | 1450*60=87000 |
| 100 | 100 | CONN | SEPERATE | 60s | 930*100=93000 |