Spring 提供了HbaseTemplate 对Hbase数据库的常规操作进行了简单的封装。

get,find方法分别对应了单行数据查询和list查询。

这些查询都要开启和关闭Hbase数据库链接


@Override
        public <T> T execute(String tableName, TableCallback<T> action) {
                Assert.notNull(action, "Callback object must not be null");
                Assert.notNull(tableName, "No table specified");

                HTableInterface table = getTable(tableName);

                try {
                        boolean previousFlushSetting = applyFlushSetting(table);
                        T result = action.doInTable(table);
                        flushIfNecessary(table, previousFlushSetting);
                        return result;
                } catch (Throwable th) {
                        if (th instanceof Error) {
                                throw ((Error) th);
                        }
                        if (th instanceof RuntimeException) {
                                throw ((RuntimeException) th);
                        }
                        throw convertHbaseAccessException((Exception) th);
                } finally {
                        releaseTable(tableName, table);
                }
        }

        private HTableInterface getTable(String tableName) {
                return HbaseUtils.getHTable(tableName, getConfiguration(), getCharset(), getTableFactory());
        }

        private void releaseTable(String tableName, HTableInterface table) {
                HbaseUtils.releaseTable(tableName, table, getTableFactory());
        }
HTableInterface table = getTable(tableName); 获取数据库链接
releaseTable(tableName, table); 释放链接

在HbaseUtils.getHTable：


if (HbaseSynchronizationManager.hasResource(tableName)) {
                        return (HTable) HbaseSynchronizationManager.getResource(tableName);
                }

看见这个大家应该都有是曾相似的感觉吧，这和Spring事务管理核心类TransactionSynchronizationManager很像，而实现也基本一样
都是通过ThreadLocal将链接保存到当前线程中。

我们要做的就是要像Srping 事务配置一样，在进入service方法时通过Aop机制将tableNames对应的链接加入到线程中。

Spring提供了这个Aop方法拦截器 HbaseInterceptor：


public Object invoke(MethodInvocation methodInvocation) throws Throwable {
                Set<String> boundTables = new LinkedHashSet<String>();
               
                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) {
                        if (this.exceptionConversionEnabled) {
                                throw convertHBaseException(ex);
                        }
                        else {
                                throw ex;
                        }
                } finally {
                        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");
                                }
                        }
                }
        }
很明显在

Object retVal = methodInvocation.proceed();
也就是我们的service方法执行前去获取Hbase链接并通过HbaseSynchronizationManager.bindResource(tableName, table);绑定到线程中。
finally中releaseTable。



Aop配置如下：

<!-- 自动扫描beans+注解功能注册 -->
        <context:component-scan base-package="com.xxx.xxx" />
        
        <!-- 根据配置文件生成hadoopConfiguration -->
        <hdp:configuration resources="classpath:/hbase-site.xml" />
        
        <!-- hadoopConfiguration == hdp:configuration -->
<!--         <hdp:hbase-configuration configuration-ref="hadoopConfiguration" /> -->

        <bean id="hbaseTemplate" class="org.springframework.data.hadoop.hbase.HbaseTemplate">
                <!-- hadoopConfiguration == hdp:configuration -->
                <property name="configuration" ref="hadoopConfiguration" />
        </bean>
        
        <bean id="hbaseInterceptor" class="org.springframework.data.hadoop.hbase.HbaseInterceptor">
                <property name="configuration" ref="hadoopConfiguration" />
                <property name="tableNames">
                        <list>
                                <value>table_name1</value>
                                <value>table_name2</value>
                        </list>
                </property>
        </bean>
        
        <!-- 使用aop增强, 织入hbase数据库链接的开启和关闭  -->
        <aop:config>
                <aop:pointcut id="allManagerMethod"
                        expression="execution(* com.xxx.xxx.*.service..*(..))" />
                <aop:advisor advice-ref="hbaseInterceptor" pointcut-ref="allManagerMethod" />
        </aop:config>

Hbase的数据库表链接跟传统数据库不太一样， 开启链接必需要表名， 所以HbaseInterceptor中必需设置private String[] tableNames;
在进入servcie方法时，tableNames中对应的表链接都会开启。这必然会造成浪费，因为并不是每个service都会把表都查询一遍。


---------------------
【转载】仅作分享，侵删
作者：whos2002110
原文：https://blog.csdn.net/whos2002110/article/details/36874389

奈斯