程式師世界 >> 編程語言 >> JAVA編程 >> JAVA綜合教程 >> 分布式事務操作之Spring+JTA+ibatis

分布式事務操作之Spring+JTA+ibatis

編輯：JAVA綜合教程

分布式事務操作之Spring+JTA+ibatis

一、技術背景

    在項目中需要操作多個數據庫，同時，需要保證操作的原子性，保證對多個數據庫的操作一致性，而傳統的JDBC事務被限定在一個單一的數據庫連接中，故引用JTA，即Java Transaction API。JTA事務比JDBC事務更強大，一個JTA事務可以有多個數據源參與者。
    現在，我們用spring+jta+ibatis來完成基本的測試框架。

二、技術實現

第一步：導入JOTM Jar包

JOTM (Java Open Transaction Manager)，一個開源JTA實現，是由ObjectWeb協會開發的功能完整的獨立的事務管理器。下載JOTM（ow2-jotm-dist-2.2.1-bin.tar.gz），解壓tar包導入lib目錄下的Jar包。或在Maven pom.xml中添加：


    org.ow2.jotm
    ow2-jotm-dist
    2.2.1
    pom
    compile

第二步：創建JotmFactoryBean

spring為JOTM提供了一個org.springframework.transaction.jta.JotmFactoryBean支持類，通過該支持類可以方便地創建JOTM本地實例。但是spring3以後spring-tx.jar 沒有JotmFactoryBean的支持，可以反編譯spring2.5的spring-tx.jar，copy新建一個類。


import javax.naming.NamingException;
import javax.transaction.SystemException;
import org.objectweb.jotm.Current;
import org.objectweb.jotm.Jotm;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;


public class JotmFactoryBean implements FactoryBean, DisposableBean {

    private Current jotmCurrent;
    private Jotm jotm;

    public JotmFactoryBean() throws NamingException {
        jotmCurrent = Current.getCurrent();
        if (jotmCurrent == null) {
            jotm = new Jotm(true, false);
            jotmCurrent = Current.getCurrent();
        }
    }

    public void setDefaultTimeout(int defaultTimeout) {
        jotmCurrent.setDefaultTimeout(defaultTimeout);
        try {
            jotmCurrent.setTransactionTimeout(defaultTimeout);
        } catch (SystemException ex) {
        }
    }

    public Jotm getJotm() {
        return jotm;
    }

    public Object getObject() {
        return jotmCurrent;
    }

    public Class getObjectType() {
        return jotmCurrent.getClass();
    }

    public boolean isSingleton() {
        return true;
    }

    public void destroy() {
        if (jotm != null)
            jotm.stop();
    }

}

第三步：創建DAO

創建基於ibatis框架的DAO接口和實現類。

IBaseDao


import java.util.List;
import java.util.Map;

import org.apache.ibatis.session.SqlSession;

/**
 * Title： 
 * Description：  Dao 接口類
 */
public interface IBaseDao {
    /**
     * 根據Id獲取對象
     * 
     * @param id 主鍵
     * @param sqlId 腳本編號
     * @return Object對象
     */
    public Object queryForObject(int id, String sqlId);

    /**
     * 根據Id獲取對象
     * 
     * @param id 主鍵
     * @param cls 返回的對象Class
     * @param sqlId 腳本編號
     * @return cls對應的類
     */
    public  T queryForObject(int id, Class cls, String sqlId);

    /**
     * 根據條件獲取對象
     * 
     * @param params  參數
     * @param sqlId 腳本編號
     * @return Object對象
     */
    public Object queryForObject(Map params, String sqlId);

    /**
     * 根據條件獲取對象
     * 
     * @param params 參數
     * @param cls 返回的對象Class
     * @param sqlId 腳本編號
     * @return cls對應的類
     */
    public  T queryForObject(Map params, Class cls, String sqlId);

    /**
     * 獲取數據總條數
     * 
     * @param params 參數
     * @param sqlId 腳本編號
     * @return 條數
     */
    public int getTotalCount(Map params, String sqlId);

    /**
     * 查詢列表
     * 
     * @param params 參數
     * @param cls 返回的對象Class
     * @param sqlId 腳本編號
     * @return 列表
     */
    public  List queryForList(Map params, Class cls, String sqlId);

    /**
     * 查詢列表
     * 
     * @param params 參數
     * @param sqlId 腳本編號
     * @return 列表
     */
    public List> queryForList(Map params, String sqlId);

    /**
     * 修改數據
     * 
     * @param object 待更新對象
     * @param sqlId 腳本編號
     * @return 條數 
     */
    public int update(Object object, String sqlId);

    /**
     * 插入數據
     * 
     * @param sqlId 腳本編號
     * @param object 待插入的對象
     * @return 條數
     */
    public int insert(Object object, String sqlId);

    /**
     * 刪除數據
     * 
     * @param id 主鍵
     * @param sqlId 腳本編號
     * @return 條數
     */
    public int delete(int id, String sqlId);

    /**
     * 刪除數據
     * 
     * @param map 待刪除的對象
     * @param sqlId  腳本編號
     * @return 條數
     */
    public int delete(Map map, String sqlId);

    /**
     * 獲取sqlSession
     * @return
     */
    public SqlSession getSqlSession();

    /**
     * 設置sqlSession
     * @param sqlSession
     */
    public void setSqlSession(SqlSession sqlSession);

}

BaseDaoImp

import java.util.List; import java.util.Map; import org.apache.ibatis.session.SqlSession; import com.ai.dimensyn.dao.common.IBaseDao; import com.ai.frame.util.ConvertUtil; /** * Title： * Description：Dao接口實現類 */ public class BaseDaoImpl implements IBaseDao { private SqlSession sqlSession; @Override public SqlSession getSqlSession() { return sqlSession; } @Override public void setSqlSession(SqlSession sqlSession) { this.sqlSession = sqlSession; } @Override public Object queryForObject(int id, String sqlId) { return getSqlSession().selectOne(sqlId, id); } @Override @SuppressWarnings("unchecked") public T queryForObject(int id, Class cls, String sqlId) { return (T) getSqlSession().selectOne(sqlId, id); } @Override public Object queryForObject(Map params, String sqlId) { return getSqlSession().selectOne(sqlId, params); } @Override @SuppressWarnings("unchecked") public T queryForObject(Map params, Class cls, String sqlId) { return (T) getSqlSession().selectOne(sqlId, params); } @Override public int getTotalCount(Map params, String sqlId) { return (Integer) getSqlSession().selectOne(sqlId, params); } @Override public List queryForList(Map params, Class cls, String sqlId) { return getSqlSession().selectList(sqlId, params); } @Override public List> queryForList(Map params, String sqlId) { List> list = getSqlSession().selectList(sqlId, params); List> beans = ConvertUtil.convertSqlMap2JavaMap(list); return beans; } @Override public int update(Object object, String sqlId) { return getSqlSession().update(sqlId, object); } @Override public int insert(Object object, String sqlId) { return (Integer) getSqlSession().insert(sqlId, object); } @Override public int delete(int id, String sqlId) { return getSqlSession().delete(sqlId, id); } @Override public int delete(Map params, String sqlId) { return getSqlSession().delete(sqlId, params); } }

IBaseMDao


import java.util.List;
import java.util.Map;


/**
 * Title： 
 * Description：  多數據源Dao接口類
 */
public interface IBaseMDao {
    /**
     * 根據Id獲取對象
     * 
     * @param id 主鍵
     * @param sqlIds 腳本編號
     * @return Object對象
     */
    public Object[] queryForObject(int id, String... sqlIds);

    /**
     * 根據Id獲取對象
     * 
     * @param id 主鍵
     * @param cls 返回的對象Class
     * @param sqlIds 腳本編號
     * @return cls對應的類
     */
    public  T[] queryForObject(int id, Class cls, String... sqlIds);

    /**
     * 根據條件獲取對象
     * 
     * @param params 參數
     * @param sqlIds 腳本編號
     * @return Object對象
     */
    public Object[] queryForObject(Map params, String... sqlIds);

    /**
     * 根據條件獲取對象
     * 
     * @param params 參數
     * @param cls 返回的對象Class
     * @param sqlIds 腳本編號
     * @return cls對應的類
     */
    public  T[] queryForObject(Map params, Class cls, String... sqlIds);

    /**
     * 獲取數據總條數
     * 
     * @param params 參數
     * @param sqlIds 腳本編號
     * @return 條數
     */
    public int[] getTotalCount(Map params, String... sqlIds);

    /**
     * 查詢列表
     * 
     * @param params 參數
     * @param cls 返回的對象Class
     * @param sqlIds 腳本編號
     * @return 列表
     */
    public  List[] queryForList(Map params, Class cls, String... sqlIds);

    /**
     * 查詢列表
     * 
     * @param params 參數
     * @param sqlIds 腳本編號
     * @return 列表
     */
    public List>[] queryForList(Map params, String... sqlIds);

    /**
     * 修改數據
     * 
     * @param object 待更新對象
     * @param sqlIds 腳本編號
     * @return 條數 
     */
    public int[] update(Object object, String... sqlIds);

    /**
     * 插入數據
     * 
     * @param object 待插入的對象
     * @param sqlIds 腳本編號
     * @return 條數
     */
    public int[] insert(Object object, String... sqlIds);

    /**
     * 刪除數據
     * 
     * @param id 主鍵
     * @param sqlIds
     * @return 條數
     */
    public int[] delete(int id, String... sqlIds);

    /**
     * 刪除數據
     * 
     * @param params 參數
     *  @param sqlIds  腳本編號
     * @return 條數
     */
    public int[] delete(Map params, String... sqlIds);
    /**
     * 先刪數據，後插入
     * @param params
     * @param delSqlId
     * @param insertSqlId
     * @return
     */
    public int[] executeDeleteInsert(Map params, String delSqlId, String insertSqlId);
}

BaseMDaoImpl


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.ResourceBundle;

import javax.transaction.UserTransaction;

import org.apache.ibatis.session.SqlSession;
import org.springframework.transaction.jta.JtaTransactionManager;


/**
 * Title： 
 * Description：多數據源Dao接口實現類
 */
public class BaseMDaoImpl implements IBaseMDao {

    private static ResourceBundle resources;

    private JtaTransactionManager transactionManager;

    private IBaseDao baseDao = new BaseDaoImpl();

    static {
        resources = ResourceBundle.getBundle("config/mybaties");
    }

    /**
     * dmcWeb SqlSession
     */
    private SqlSession dmcWebSession;

    /**
     * dmcWeb DataSource key
     */
    private String dmcWebSrcKey;
    /**
     * rpt_web SqlSession
     */
    private SqlSession rptWebSession;

    /**
     * rpt_web DataSource key
    */
    private String rptWebSrcKey;

    /**
     * idxWeb SqlSession
     */
    private SqlSession idxWebSession;

    /**
     * idxWeb DataSource key
     */
    private String idxWebSrcKey;

    /**
     * default DataSource key
     */
    private String defaultSrcKey;

    public JtaTransactionManager getTransactionManager() {
        return transactionManager;
    }

    public void setTransactionManager(JtaTransactionManager transactionManager) {
        this.transactionManager = transactionManager;
    }

    public SqlSession getDmcWebSession() {
        return dmcWebSession;
    }


    public void setDmcWebSession(SqlSession dmcWebSession) {
        this.dmcWebSession = dmcWebSession;
    }


    public String getDmcWebSrcKey() {
        return dmcWebSrcKey;
    }


    public void setDmcWebSrcKey(String dmcWebSrcKey) {
        this.dmcWebSrcKey = dmcWebSrcKey;
    }


    public SqlSession getIdxWebSession() {
        return idxWebSession;
    }


    public void setIdxWebSession(SqlSession idxWebSession) {
        this.idxWebSession = idxWebSession;
    }


    public String getIdxWebSrcKey() {
        return idxWebSrcKey;
    }


    public void setIdxWebSrcKey(String idxWebSrcKey) {
        this.idxWebSrcKey = idxWebSrcKey;
    }

    public SqlSession getRptWebSession() {
        return rptWebSession;
    }

    public void setRptWebSession(SqlSession rptWebSession) {
        this.rptWebSession = rptWebSession;
    }

    public String getRptWebSrcKey() {
        return rptWebSrcKey;
    }

    public void setRptWebSrcKey(String rptWebSrcKey) {
        this.rptWebSrcKey = rptWebSrcKey;
    }

    public String getDefaultSrcKey() {
        return defaultSrcKey;
    }

    public void setDefaultSrcKey(String defaultSrcKey) {
        this.defaultSrcKey = defaultSrcKey;
    }

    /**
     * 獲取sqlSession
     * @param sqlId
     * @return
     */
    private SqlSession getSqlSession(String sqlId) {
        String srcKey = null;

        try {
            srcKey = resources.getString(sqlId);
        } catch (Exception e) {

        }

        if (null == srcKey) {
            srcKey = defaultSrcKey;
        }

        if (dmcWebSrcKey.equals(srcKey)) {
            return dmcWebSession;
        } else if (idxWebSrcKey.equals(srcKey)) {
            return idxWebSession;
        } else if (rptWebSrcKey.equals(srcKey)) {
            return rptWebSession;
        } else {
            throw new NullPointerException("SqlSession null: " + sqlId);
        }
    }

    @Override
    public Object[] queryForObject(int id, String... sqlIds) {
        Object[] results = new Object[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.queryForObject(id, sqlIds[i]);
        }
        return results;
    }

    @Override
    @SuppressWarnings("unchecked")
    public  T[] queryForObject(int id, Class cls, String... sqlIds) {
        T[] results = (T[])new Object[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.queryForObject(id, cls, sqlIds[i]);
        }
        return results;
    }

    @Override
    public Object[] queryForObject(Map params, String... sqlIds) {
        Object[] results = new Object[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.queryForObject(params, sqlIds[i]);
        }
        return results;
    }

    @Override
    @SuppressWarnings("unchecked")
    public  T[] queryForObject(Map params, Class cls, String... sqlIds) {
        T[] results = (T[])new Object[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.queryForObject(params, cls, sqlIds[i]);
        }
        return results;
    }

    @Override
    public int[] getTotalCount(Map params, String... sqlIds) {
        int[] results = new int[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.getTotalCount(params, sqlIds[i]);
        }
        return results;
    }

    @Override
    @SuppressWarnings("unchecked")
    public  List[] queryForList(Map params, Class cls, String... sqlIds) {
        List[] results = new ArrayList[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            results[i] = baseDao.queryForList(params, cls, sqlIds[i]);
        }
        return results;
    }

    @Override
    @SuppressWarnings("unchecked")
    public List>[] queryForList(Map params, String... sqlIds) {
        List>[] results = new ArrayList[sqlIds.length];
        for (int i = 0; i < sqlIds.length; i++) {
            baseDao.setSqlSession(getSqlSession(sqlIds[i]));
            //單獨改造，滿足根據 table&方法 共同決定切換數據源
            System.out.println(sqlIds[i].substring(0, sqlIds[i].indexOf("-")));
            results[i] = baseDao.queryForList(params, sqlIds[i].substring(0, sqlIds[i].indexOf("-")));
        }
        return results;
    }

    @Override
    public int[] update(Object object, String... sqlIds) {
        int[] results = new int[sqlIds.length];
        UserTransaction transaction = this.transactionManager.getUserTransaction();
        try {
            transaction.begin();
            for (int i = 0; i < sqlIds.length; i++) {
                baseDao.setSqlSession(getSqlSession(sqlIds[i]));
                results[i] = baseDao.update(object, sqlIds[i]);
            }
            transaction.commit();
        } catch (Exception e1) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                throw new RuntimeException(e2);
            }
            throw new RuntimeException(e1);
        }
        return results;
    }

    @Override
    public int[] insert(Object object, String... sqlIds) {
        int[] results = new int[sqlIds.length];
        UserTransaction transaction = this.transactionManager.getUserTransaction();
        try {
            transaction.begin();
            for (int i = 0; i < sqlIds.length; i++) {
                baseDao.setSqlSession(getSqlSession(sqlIds[i]));
                results[i] = baseDao.insert(object, sqlIds[i]);
            }
            transaction.commit();
        } catch (Exception e1) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                throw new RuntimeException(e2);
            }
            throw new RuntimeException(e1);
        }
        return results;
    }

    @Override
    public int[] delete(int id, String... sqlIds) {
        int[] results = new int[sqlIds.length];
        UserTransaction transaction = this.transactionManager.getUserTransaction();
        try {
            transaction.begin();
            for (int i = 0; i < sqlIds.length; i++) {
                baseDao.setSqlSession(getSqlSession(sqlIds[i]));
                results[i] = baseDao.delete(id, sqlIds[i]);
            }
            transaction.commit();
        } catch (Exception e1) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                throw new RuntimeException(e2);
            }
            throw new RuntimeException(e1);
        }
        return results;
    }

    @Override
    public int[] delete(Map params, String... sqlIds) {
        int[] results = new int[sqlIds.length];
        UserTransaction transaction = this.transactionManager.getUserTransaction();
        try {
            transaction.begin();
            for (int i = 0; i < sqlIds.length; i++) {
                baseDao.setSqlSession(getSqlSession(sqlIds[i]));
                results[i] = baseDao.delete(params, sqlIds[i]);
            }
            transaction.commit();
        } catch (Exception e1) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                throw new RuntimeException(e2);
            }
            throw new RuntimeException(e1);
        }
        return results;
    }

    @Override
    public int[] executeDeleteInsert(Map params, String delSqlId, String insertSqlId) {
        UserTransaction transaction = this.transactionManager.getUserTransaction();
        try {
            transaction.begin();
            baseDao.setSqlSession(getSqlSession(delSqlId));
            baseDao.delete(params, delSqlId.substring(0, delSqlId.indexOf("-")));
            baseDao.setSqlSession(getSqlSession(insertSqlId));
            //對sql進行；分隔 迭代插入
            String[] strsOfSql = params.get(Constants.SQL).split(";");
            Map sqlMap = new HashMap();
            for (String sql : strsOfSql) {
                sqlMap.put(Constants.SQL, sql);
                baseDao.insert(sqlMap, insertSqlId.substring(0,insertSqlId.indexOf("-")));
                sqlMap.clear();
            }
            transaction.commit();
        } catch (Exception e1) {
            try {
                transaction.rollback();
            } catch (Exception e2) {
                throw new RuntimeException(e2);
            }
            throw new RuntimeException(e1);
        }
        return null;
    }

}

第四步：XML配置

SpringFactory：

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;


public class SpringFactory implements BeanFactoryAware {

    private static BeanFactory beanFactory;  

    // private static ApplicationContext context;  

    public void setBeanFactory(BeanFactory factory) throws BeansException {  
        this.beanFactory = factory;  
    }  

    /** 
     * 根據beanName名字取得bean 
     *  
     * @param beanName 
     * @return 
     */  
    public static  T getBean(String beanName) {  
        if (null != beanFactory) {  
            return (T) beanFactory.getBean(beanName);  
        }  
        return null;  
    }  

}

第五步：測試

在後台測試分布式事務管理：


private IBaseMDao baseMDao = SpringFactory.getBean("baseMDao");

public void testBaseMDao() {
    try {
        Map params = new HashMap();
        params.put("table_tb",table);
        params.put(Constants.SQL,sql);
        baseMDao.executeDeleteInsert(params,"dimensyn.deleteDimenTable"+"-"+table, "dimensyn.insertDimenTable"+"-"+table);

    } catch (Exception e) {
        e.printStackTrace();
    }
}

public IBaseMDao getBaseMDao() {
    return baseMDao;
}

public void setBaseMDao(IBaseMDao baseMDao) {
    this.baseMDao = baseMDao;
}

在indexWeb、dmcWeb庫中創建測試表，並插入測試數據，配置：


    

    

    
        delete from ${table_tb}

Configuration.xml



    
    
       
    


mybaties.properties：
dimensyn.deleteDimenTable-testtable_xuchun=indexWeb
dimensyn.insertDimenTable-testtable_xuchun=dmcWeb