【Java EE 学习 77 下】【数据采集系统第九天】【使用spring实现答案水平分库】【未解决问题：分库查询问题】

　　之前说过，如果一个数据库中要存储的数据量整体比较小，但是其中一个表存储的数据比较多，比如日志表，这时候就要考虑分表存储了；但是如果一个数据库整体存储的容量就比较大，该怎么办呢？这时候就需要考虑分库了，就是建立多个数据库保存数据。这里以答案为例，就算调查对象不是很多，但是参与调查的人数非常多，那么需要保存的数据量就会非常大，怎样将答案以一种规则保存到不同的数据库中就是现在需要考虑的问题（查询分库的问题未解决，先存档）。

一、分库方法

　　分库分为水平分库和竖直分库两种类型。

　　（1）水平分库

　　　　数据库之间是同构的，但是数据的存储范围不同。比如之后我将使用水平分库的方法保存答案到不同的数据库中。两个数据库中都有答案表，而且字段和约束等完全相同，两者的差异只是保存的数据不同，这样的分库方法就是水平分库。

　　（2）竖直分库

　　　　数据库和数据库之间的结构不相同，比如一个数据库存放一个模块的功能，每个模块的独立性比较强。而且量比较大。

二、实现答案分库的步骤：以2个数据库为例说明

　　1.创建第二个数据库lsn_surveypark1

　　2.配置数据源

　　　　因为之前已经配置过数据源了，所以这里只需要直接继承上一个数据源并且修改url地址即可

 <!-- 配置数据源（主库） -->
 <bean id="dateSource" class="com.mchange.v2.c3p0.ComboPooledDataSource">
     <property name="driverClass" value="${jdbc.driverclass}"></property>
     <property name="jdbcUrl" value="${jdbc.url}"></property>
     <property name="user" value="${jdbc.username}"></property>
     <property name="password" value="${jdbc.password}"></property>

     <!-- 配置c3p0自身的参数 -->
     <property name="maxPoolSize" value="${c3p0.pool.maxsize}"></property>
     <property name="minPoolSize" value="${c3p0.pool.minsize}"></property>
     <property name="initialPoolSize" value="${c3p0.pool.initsieze}"></property>
     <property name="acquireIncrement" value="${c3p0.pool.increment}"></property>
 </bean>
 <!-- （从库） 为了实现分库的功能，必须针对每个数据库配置一个数据源 这里使用了包的继承的特殊属性使用parent属性对dataSource进行了继承 -->
 <bean id="dataSource1" class="com.mchange.v2.c3p0.ComboPooledDataSource"
     parent="dateSource">
     <property name="jdbcUrl" value="jdbc:mysql://localhost:3306/lsn_surveypark1"></property>
 </bean>

　　3.配置数据源路由器

　　　　数据源路由器将会根据策略决定使用的数据源。

 <!-- 配置数据源路由器 -->
 <bean id="dataSource_router" class="com.kdyzm.datasource.SurveyparkDatasourceRouter">
     <property name="targetDataSources">
         <map>
             <!-- 如果id是偶数，保存到主库中 -->
             <entry key="even" value-ref="dateSource"></entry>
             <!-- 如果id是奇数，保存到从库中 -->
             <entry key="odd" value-ref="dataSource1"></entry>
         </map>
     </property>
     <!-- 如果不满足上述规则，则直接使用默认的数据源 -->
     <property name="defaultTargetDataSource" ref="dateSource"></property>
 </bean>

　　这里的策略封装到了一个类中SurveyparkDatasourceRouter，该类必须继承org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource抽象类并重写determineCurrentLookupKey方法确定策略。

　　4.自定义路由数据源策略

　　自定义的方法就是继承org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource类并重写抽象方法。

 package com.kdyzm.datasource;

 import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;

 import com.kdyzm.domain.Survey;

 /**
  * 自定义数据源路由器
  *     有一个默认的实现类，该类是以传播属性来路由数据的。
  * @author kdyzm
  *
  */
 public class SurveyparkDatasourceRouter extends AbstractRoutingDataSource{

     /**
      * 该方法实际上确定了一个数据向哪里存放的策略
      * 在这里使用id的就属性来确定
      * 如果答案id是偶数，就想lsn_surveypark数据库中的answer表（主表）中存放
      * 如果答案的id是奇数，就向lsn_surveypark1数据库中的answer表（从表）中存放
      */
     @Override
     protected Object determineCurrentLookupKey() {
         SurveyToken surveyToken=SurveyToken.getSurveyToken();
         if(surveyToken!=null){
             Survey survey=surveyToken.getSurvey();
             int surveyId=survey.getSurveyId();
             System.out.println("Survey对象不为空，值为："+surveyId);
             /**
              * 在这里必须解除绑定
              * 如果不在这里解除绑定的话就会将log日志写入到lsn_surveypark1数据库中。
              * 由于lsn_surveypark1数据库中没有log表，所以一定会报错
              */
             SurveyToken.unbind();
             return (surveyId%2)==0?"even":"odd";    //如果是偶数返回even字符串，如果是奇数返回odd字符串
         }
         System.out.println("survey对象为空");
         return null;
     }

 }

　　以上重写的方法中决定了路由数据源的策略：如果调查ID是偶数，就保存到主库lsn_surveypark的answer表中，如果是奇数，就保存到从库lsn_surveypark1中的answer表中。

　　接下来就是解决怎么拿到Survey对象的问题，上面的黄色背景部分的代码是关键。

　　注意，如果Survey对象为空，就使用默认的数据源：主库，这是由之前的配置文件中的配置决定的。

　　5.首先解决执行顺序问题

　　　　什么时候决定数据源？这个问题不太确定，应该是在进入Service方法之前，也就是说开启事务的时候（在分库查询的时候这种猜测被推翻了）。那么只要在进入Service方法之前将Survey对象传递给路由数据源中的相关方法就行了。

　　6.使用ThreadLocal解决拿到Survey的问题。

　　（1）分析问题

　　　　保存问题的时机是SurveyAction调用保存答案方法的时候。这时候就需要将数据保存到某个地方然后等待在determineCurrentLookupKey方法中获取该值就可以了。但是保存到哪里呢？保存到文件中是一种方式，但是这种方式非常烂~通常这种情况下都是讲对象绑定到ThreadLocal，然后在determineCurrentLookupKey方法中从ThreadLocal中拿出来即可。

　　　　我在这里创建一个新类SurveyToken，实现设置Survey对象、获取Survey对象的、将Survey对象绑定到ThreadLocal（实际上是当前线程）和将Survey对象从ThreadLocal解除绑定的方法，当然前两者是非静态方法，后两者是静态方法。

 package com.kdyzm.datasource;

 import com.kdyzm.domain.Survey;

 /**
  * 令牌类
  * 封装了一些比较重要的属性
  * @author kdyzm
  *
  */
 public class SurveyToken {
     private Survey survey;        //绑定的对象的值，如果只是绑定surveyId也可以，但是为了以后的方便起见，使用该对象更划算
     private static ThreadLocal<SurveyToken> t=new ThreadLocal<SurveyToken>();
     public Survey getSurvey() {
         return survey;
     }
     public void setSurvey(Survey survey) {
         this.survey = survey;
     }
     /**
      * 绑定当前线程和SurveyToken对象之间的关系
      */
     public static void bind(SurveyToken surveyToken){
         t.set(surveyToken);
     }

     /**
      * 解除当前线程和SurveyToken对象之间的关系
      */
     public static void unbind(){
         t.remove();
     }

     /**
      * 获取SurveyToken对象的方法
      */
     public static SurveyToken getSurveyToken(){
         return t.get();
     }
 }

　　（2）在EntrySurveyAction中调用Service方法之前绑定Survey对象到ThreadLocal

     private void writeAnswersToDB(List<Answer> answers) {
         SurveyToken surveyToken=new SurveyToken();
         Survey survey=this.surveyService.getModelById(getSurveyId());
         surveyToken.setSurvey(survey);
         SurveyToken.bind(surveyToken);
         this.answerService.saveAllAnswers(answers);
     }

　　　　因为answerService类中的saveAllAnswers方法带有事务，所以在调用该方法之前会调用determineCurrentLookupKey方法决定数据源。

　　7.测试

　　　　如果只是经过了以上几个步骤，测试一定是失败的。

　　　　应该会报出"在lsn_surveypark1数据库中无法找到log表"诸如此类的异常信息。

　　　　分析原因：lsn_surveypark1数据库本来就是从数据库，里面只有一张answer表，本来就没有log表，说明程序选取的数据源有问题。要知道，只要事务没结束，determineCurrentLookupKey方法就不会有机会被再次调用，即使中间可能会再次调用其它Service中的方法也没用，因为事务的传播性为"REQUIRED"，这样就导致了其调用的所有方法都自动开启了事务，当然"保存日志"的动作也是"其它Service"中的方法，当然也就不会重新访问determineCurrentLookupKey方法，数据源也就会一直是lsn_surveypark1，因此就报出了上述的那个错误。所以就找到了问题的关键：事务通知和日志通知的开启顺序导致了该错误的发生，我们需要让事务通知在后，日志通知在前，所以在配置AOP的时候就需要改变order属性，使得日志通知的order值小于事务通知的order值，这样就会先开启日志通知，再开启事务通知了，这样做的结果就是一旦保存答案完成之后，保存答案的事务就会结束；日志通知就会为了保存日志再次访问determineCurrentLookupKey方法，当然这时候必须保证Survey对象已经解除了绑定，否则仍然会使用之前确定的数据源，所以解除绑定的时机也很重要，如果在Action中解除绑定，即使颠倒了事务通知和日志通知的启动顺序也没有什么作用，最好的方法就是在determineCurrentLookupKey方法中拿到Survey对象之后直接解除，这样就能够保证一次事务结束之后下一次事务开启的时候访问determineCurrentLookupKey的时候Survey对象已经解除绑定了。配置事务通知和日志通知的顺序方法如下：

 <aop:config>
     <!-- 日志切入点 -->
     <aop:pointcut
         expression="(execution(* *..*Service.save*(..))
                                 or execution(* *..*Service.update*(..))
                                 or execution(* *..*Service.delete*(..))
                                 or execution(* *..*Service.batch*(..))
                                 or execution(* *..*Service.create*(..))
                                 or execution(* *..*Service.new*(..))) and !bean(logService)"
         id="loggerPointcut" />
     <aop:pointcut expression="execution(* *..*Service.*(..))"
         id="txPointcut" />
     <!-- 必须配置order属性，使用该属性可以改变配置的通知的加载顺序，order值越大，优先级越高 必须让事务的通知放到后面，让日志的通知先执行，这样才能在执行完成日志的通知后事务确保能够结束。
         order值越小，优先级越高 为了解决事务没有结束的问题，必须同时修改解除绑定的时间 -->
     <aop:advisor advice-ref="cacheAdvice"
         pointcut="execution(* com.kdyzm.service.SurveyService.*(..)) or
                 execution(* com.kdyzm.service.PageService.*(..)) or
                 execution(* com.kdyzm.service.QuestionService.*(..)) or
                 execution(* com.kdyzm.service.AnswerService.*(..))" order="0" />
     <aop:advisor advice-ref="txAdvice" pointcut-ref="txPointcut"
        order="2" />
     <aop:aspect id="loggerAspect" ref="logger" order="1">
         <aop:around method="record" pointcut-ref="loggerPointcut" />
     </aop:aspect>
 </aop:config>

　　　　注意不要忘了在determineCurrentLookupKey方法中拿到Survey对象之后直接解除绑定，如果在Action中解除绑定的话，就算颠倒日志通知和事务通知的启动顺序也是没有任何作用的。