脫離 Spring 實現複雜嵌套事務，之十（實現篇）

    本文是<實現 Spring 的事務控制>系列文章中一篇。本文假設讀者已經閱讀並理解《實現 Spring 的事務控制，之九（整合七種傳播行爲）》。其實本文大多數時間是在貼代碼。

    事務控制的完整代碼位於：http://git.oschina.net/zycgit/hasor，DB模塊中。

    在開始實現事務管理器以前，咱們須要知道。這個系統中都有那些元素？

數據庫鏈接：

    數據庫鏈接是封裝的最基本的元素，咱們上文中提到的引用計數能夠封裝在這個對象上。

資源管理器：

    首先程序不能直接經過 DataSource 取得數據庫鏈接，而是必須經過資源管理器去申請。它有一些充當數據庫鏈接池的角色。可是請你們千萬要和數據庫鏈接池區分開，資源管理器在這裏的主要做用是管理當前鏈接。

事務狀態：

    事務狀態是封裝了「new」、「suspent」、「savepoint」等狀態的封裝對象，每當從事務管理器中開啓一個行爲的事務以後。事務管理器都會建立一個事務狀態用以表示。當遞交事務的時候也須要程序提供相應的狀態，告訴事務管理器執行那個點上的事務操做。

事務管理器：

    一切元素的控制中心，它就是核心部件。專門用來實現事務控制的。

---------------------------------------------------------------------------------

構建程序邏輯框架：

public class DefaultTransactionManager implements TransactionManager {
    /**開啓事務*/
    public final TransactionStatus getTransaction(TransactionBehavior behavior) throws SQLException {
        return getTransaction(behavior, TransactionLevel.ISOLATION_DEFAULT);
    };
    /**開啓事務*/
    public final TransactionStatus getTransaction(TransactionBehavior behavior, TransactionLevel level) throws SQLException {
        ......
    }
    /**判斷鏈接對象是否處於事務中，該方法會用於評估事務傳播屬性的處理方式。 */
    private boolean isExistingTransaction(TransactionStatus defStatus) throws SQLException {
        ......
    };
    /**初始化一個新的鏈接，並開啓事務。*/
    protected void doBegin(TransactionStatus defStatus) throws SQLException {
        ......
    }
    /**遞交事務*/
    public void commit(TransactionStatus status) throws SQLException {
        ......
    }
    /**回滾事務*/
    public void rollBack(TransactionStatus status) throws SQLException {
        ......
    }
    //
    /**掛起事務。*/
    protected void suspend(DefaultTransactionStatus defStatus) {
        ......
    }
    /**恢復被掛起的事務。*/
    protected void resume(DefaultTransactionStatus defStatus) {
        ......
    }
    /**commit,rollback。以後的清理工做，同時也負責恢復事務和操做事務堆棧。*/
    private void cleanupAfterCompletion(DefaultTransactionStatus defStatus) {

    }
    /**獲取數據庫鏈接（線程綁定的）*/
    protected TransactionObject doGetConnection(DefaultTransactionStatus defStatus) {
        ......
    };
    /**釋放數據庫鏈接*/
    protected void releaseConnection(DefaultTransactionStatus defStatus) {
        ......
    }
}

getTransaction方法

    這個方法是咱們用於處理開啓事務的方法，在上一篇文章中開啓事務方面的程序都是從這裏開始寫的。當建立好事務應當返回一個事務狀態對象用 TransactionStatus 接口去封裝。

isExistingTransaction方法

   這個方法是專門用於判斷數據庫鏈接是否存在事務的。咱們能夠想象，它必定會和 Connection 打交道的。在前面幾篇文章中我已經介紹過，幾乎每種傳播行爲都會去判斷是否存在事務。所以單獨分離出這個方法也比較有用。

doBegin方法

    處理開啓事務的操做，還記得new狀態把。凡是沒有開啓事務的數據庫鏈接只要開啓了事務都是知足了new狀態的。所以設置new狀態的代碼就能夠在這裏去實現它。

commit方法、rollBack方法

   事務的遞交和回滾操做入口，在這兩個方法裏會去判斷那一大堆狀態以及如何進行遞交或回滾的操做。

suspend方法、resume方法

    掛起和恢復，掛起方法主要作的事情就是將當前數據庫鏈接清除。而後從新申請一個鏈接做爲當前鏈接，恢復方法是將一個數據庫鏈接從新做爲當前鏈接。

cleanupAfterCompletion方法

    負責清理事務狀態的方法。

doGetConnection方法

    負責申請數據庫鏈接，並處理引用計數+1。

releaseConnection方法

    負責釋放鏈接，引用計數-1。

---------------------------------------------------------------------------------

下面這個是getTransaction方法的實現：

/**開啓事務*/
    public final TransactionStatus getTransaction(TransactionBehavior behavior, TransactionLevel level) throws SQLException {
        Hasor.assertIsNotNull(behavior);
        Hasor.assertIsNotNull(level);
        //1.獲取鏈接
        DefaultTransactionStatus defStatus = new DefaultTransactionStatus(behavior, level);
        defStatus.setTranConn(doGetConnection(defStatus));
        /*-------------------------------------------------------------
        |                      環境已經存在事務
        |
        | PROPAGATION_REQUIRED     ：加入已有事務（不處理）
        | RROPAGATION_REQUIRES_NEW ：獨立事務（掛起當前事務，開啓新事務）
        | PROPAGATION_NESTED       ：嵌套事務（設置保存點）
        | PROPAGATION_SUPPORTS     ：跟隨環境（不處理）
        | PROPAGATION_NOT_SUPPORTED：非事務方式（僅掛起當前事務）
        | PROPAGATION_NEVER        ：排除事務（異常）
        | PROPAGATION_MANDATORY    ：強制要求事務（不處理）
        ===============================================================*/
        if (this.isExistingTransaction(defStatus) == true) {
            /*RROPAGATION_REQUIRES_NEW：獨立事務*/
            if (behavior == RROPAGATION_REQUIRES_NEW) {
                this.suspend(defStatus);/*掛起當前事務*/
                this.doBegin(defStatus);/*開啓新事務*/
            }
            /*PROPAGATION_NESTED：嵌套事務*/
            if (behavior == PROPAGATION_NESTED) {
                defStatus.markHeldSavepoint();/*設置保存點*/
            }
            /*PROPAGATION_NOT_SUPPORTED：非事務方式*/
            if (behavior == PROPAGATION_NOT_SUPPORTED) {
                this.suspend(defStatus);/*掛起事務*/
            }
            /*PROPAGATION_NEVER：排除事務*/
            if (behavior == PROPAGATION_NEVER) {
                throw new IllegalTransactionStateException("Existing transaction found for transaction marked with propagation 'never'");
            }
            return pushStack(defStatus);/*入棧*/
        }
        /*-------------------------------------------------------------
        |                      環境不經存在事務
        |
        | PROPAGATION_REQUIRED     ：加入已有事務（開啓新事務）
        | RROPAGATION_REQUIRES_NEW ：獨立事務（開啓新事務）
        | PROPAGATION_NESTED       ：嵌套事務（開啓新事務）
        | PROPAGATION_SUPPORTS     ：跟隨環境（不處理）
        | PROPAGATION_NOT_SUPPORTED：非事務方式（不處理）
        | PROPAGATION_NEVER        ：排除事務（不處理）
        | PROPAGATION_MANDATORY    ：強制要求事務（異常）
        ===============================================================*/
        /*PROPAGATION_REQUIRED：加入已有事務*/
        if (behavior == PROPAGATION_REQUIRED ||
        /*RROPAGATION_REQUIRES_NEW：獨立事務*/
        behavior == RROPAGATION_REQUIRES_NEW ||
        /*PROPAGATION_NESTED：嵌套事務*/
        behavior == PROPAGATION_NESTED) {
            this.doBegin(defStatus);/*開啓新事務*/
        }
        /*PROPAGATION_MANDATORY：強制要求事務*/
        if (behavior == PROPAGATION_MANDATORY) {
            throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");
        }
        //
        return pushStack(defStatus);/*入棧*/
    }

有興趣的同窗能夠逐行分析，該方法的實現原理是依照《整合七種傳播行爲》一文。

下面這段代碼展現瞭如何封裝數據庫鏈接，包括瞭如何處理引用計數：

public class ConnectionHolder implements SavepointManager {
    private int        referenceCount;
    private DataSource dataSource;
    private Connection connection;
    public ConnectionHolder(DataSource dataSource) {
        this.dataSource = dataSource;
    }
    /**增長引用計數,一個由於持有人已被請求。*/
    public synchronized void requested() {
        this.referenceCount++;
    }
    /**減小引用計數,一個由於持有人已被釋放。*/
    public synchronized void released() {
        this.referenceCount--;
        if (!isOpen() && this.connection != null) {
            try {
                this.savepointCounter = 0;
                this.savepointsSupported = null;
                this.connection.close();
            } catch (SQLException e) {
                throw new DataAccessException("cant not close connection.", e);
            } finally {
                this.connection = null;
            }
        }
    }
    public boolean isOpen() {
        if (referenceCount == 0)
            return false;
        return true;
    }
    /**獲取鏈接*/
    public synchronized Connection getConnection() throws SQLException {
        if (this.isOpen() == false)
            return null;
        if (this.connection == null) {
            this.connection = this.dataSource.getConnection();
        }
        return this.connection;
    }
    /**是否存在事務*/
    public boolean hasTransaction() throws SQLException {
        Connection conn = getConnection();
        if (conn == null)
            return false;
        //AutoCommit被標記爲 false 表示開啓了事務。
        return conn.getAutoCommit() == false ? true : false;
    }
}

接下來這段代碼展現瞭如何遞交事務，回滾事務的處理方式亦是如此：

public final void commit(TransactionStatus status) throws SQLException {
        DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
        /*已完畢，不須要處理*/
        if (defStatus.isCompleted())
            throw new IllegalTransactionStateException("Transaction is already completed - do not call commit or rollback more than once per transaction");
        /*-------------------------------------------------------------
        | 1.不管何種傳播形式，遞交事務操做都會將 isCompleted 屬性置爲 true。
        | 2.若是事務狀態中包含一個未處理的保存點。僅遞交保存點，而非遞交整個事務。
        | 3.事務 isNew 只有爲 true 時才真正觸發遞交事務操做。
        ===============================================================*/
        try {
            prepareCommit(defStatus);
            if (defStatus.isReadOnly() || defStatus.isRollbackOnly()) {
                /*回滾狀況*/
                if (Hasor.isDebugLogger())
                    Hasor.logDebug("Transactional code has requested rollback");
                doRollback(defStatus);
            } else {
                /*若是包含保存點，在遞交事務時只處理保存點*/
                if (defStatus.hasSavepoint())
                    defStatus.releaseHeldSavepoint();
                else if (defStatus.isNewConnection())
                    doCommit(defStatus);
            }
            //
        } catch (SQLException ex) {
            doRollback(defStatus);/*遞交失敗，回滾*/
            throw ex;
        } finally {
            cleanupAfterCompletion(defStatus);
        }
    }