spring hadoop系列(四) --- spring hadoop hbase

天成1

一、源码分析
/**
*
* hbase 采用aop的拦截器，将hbase table绑定对应的线程在被调用之前；
* 一旦对应的调用方法获取结果之后将关闭并移除
* 如果有一个已经绑定的table（来自前面的call或者transaction），
* 那么 interceptor 只是简单的操作
* 同时interceptor经常与HbaseSynchronizationManager结合使用
* @author Costin Leau
*/
public class HbaseInterceptor extends HbaseAccessor implements MethodInterceptor {

private static final Log log = LogFactory.getLog(HbaseInterceptor.class);
        // 是否进行异常转换(1、输出异常链信息 2、输出原始异常信息)
private boolean exceptionConversionEnabled = true;
        // hbase要进行绑定的table name
private String[] tableNames;

        // 判断table name 集合是否为空
public void afterPropertiesSet() {
super.afterPropertiesSet();
Assert.notEmpty(tableNames, "at least one table needs to be specified");
}
   
public Object invoke(MethodInvocation methodInvocation) throws Throwable {
                // 存放table name 与 hbase table 关联映射
Set<String> boundTables = new LinkedHashSet<String>();
                // 通过HbaseSynchronizationManager判断对应的table name
                // 是否已经绑定
                // 若是没有绑定 则根据tableName、hbase Configuration
                // 、charset、table factory构建HTableInterface对象
                // 完成绑定之后，执行方法调用；并返回处理结果
for (String tableName : tableNames) {
if (!HbaseSynchronizationManager.hasResource(tableName)) {
boundTables.add(tableName);
HTableInterface table = HbaseUtils.getHTable(tableName, getConfiguration(), getCharset(), getTableFactory());
HbaseSynchronizationManager.bindResource(tableName, table);
}
}

try {
Object retVal = methodInvocation.proceed();
return retVal;
} catch (Exception ex) {
                        // 根据exceptionConversionEnabled的内容
                        // 输出异常信息链
if (this.exceptionConversionEnabled) {
throw convertHBaseException(ex);
}
else {
                        // 若是exceptionConversionEnabled为false
                        // 则只输出原始异常
throw ex;
}
} finally {
                        // 方法调用完成之后 要将前面绑定的table
                        // 执行release操作
for (String tableName : boundTables) {
HTableInterface table = (HTableInterface) HbaseSynchronizationManager.unbindResourceIfPossible(tableName);
if (table != null) {
HbaseUtils.releaseTable(tableName, table);
}
else {
log.warn("Table [" + tableName + "] unbound from the thread by somebody else; cannot guarantee proper clean-up");
}
}
}
}
        //
private Exception convertHBaseException(Exception ex) {
return HbaseUtils.convertHbaseException(ex);
}

public void setTableNames(String[] tableNames) {
this.tableNames = tableNames;
}

/**
* Sets whether to convert any {@link IOException} raised to a Spring DataAccessException,
* compatible with the <code>org.springframework.dao</code> exception hierarchy.
* <p>Default is "true". Turn this flag off to let the caller receive raw exceptions
* as-is, without any wrapping.
* @see org.springframework.dao.DataAccessException
*
* @param exceptionConversionEnabled enable exceptionConversion
*/
public void setExceptionConversionEnabled(boolean exceptionConversionEnabled) {
this.exceptionConversionEnabled = exceptionConversionEnabled;
}
}
该类主要使用hbase table之前完成bind的操作；完成table name到具体的hbase table之间的映射；不过在执行完成对应的操作之后来释放绑定的资源