.NET簡談事務、分布式事務處理

 

 

在本人的 “ .NET簡談事務本質論”一文中我們從整體上了解了事務模型，在我們腦子里能有一個全局的事務處理結構，消除對數據庫事務的依賴理解，重新認識事務編程模型。

今天這篇文章我們將使用.NET C#來進行事務性編程，從淺顯、簡單的本地事務開始，也就是我們用的最多的ADO.NET事務處理，然后我們逐漸擴大事務處理范圍，包括對分布式事務處理的使用，多線程事務處理的使用。

 

數據庫事務處理

數據庫事務處理我們基本都很熟悉了，begin Transaction ……end Transaction，將要進行事務性的操作包在代碼段里，為了便於文章有條理的講解下去，我還是在這里穿插一個簡單的小示例，便於與后面的代碼進行對比分析。

例 1 ：

我們在數據庫里建兩張表，也就是很簡單一列信息。

表1名：test

 

表2名：test2

 

目的是為了掩飾事務的特性，所以我們這里給表1test的name列設置為主鍵，我們后面將通過有意的造成主鍵重復，導致事務自動回滾的效果。

我先來解釋一下這兩張表后面干什么用的。表test是用來有意造成事務內部處理出錯用的，表test2是用來在事務處理當中扮演着沒有錯誤的常規數據插入用的，我會在test2中先插入數據，然后在test中插入數據時觸發事務內部執行錯誤導致事務回滾。

好了我們進行T-SQL的編寫：

 

insert into test values('222') --我們在表test中插入一條記錄
go
begin transaction tr
begin try
begin
insert into test2 values('111')
insert into test values('222') --該行插入會導致主鍵沖突，也就是我們要的效果
end
commit transaction tr
end try
begin catch
print '事務執行錯誤!'
print error_number()
rollback transaction tr
end catch

我們運行看看結果：

在事務處理過程中，很明顯第一條插入語句執行成功了，但是由於第二條插入語句導致事務回滾所以數據是沒有變化的。

這個示例可能過於簡單，真正的企業級應用可能很復雜，但是我們的目的是看看事務的使用，越簡單越明了越好。 [王清培版權所有，轉載請給出署名]

 

ADO.NET 事務處理

下面我們將事務在.NET的AOD.NET中實現看看效果。

例2：

public class Test
    {
        SqlConnection conn = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        public void Add()
        {
            conn.Open();
            SqlCommand command = new SqlCommand("insert into test2 values(111)", conn);
            try
            {
                command.Transaction = conn.BeginTransaction();
                command.ExecuteNonQuery();
                command.CommandText = "insert into test values(222)";
                command.ExecuteNonQuery();
                command.Transaction.Commit();
            }
            catch (Exception err)
            {
                Console.WriteLine(err);
                command.Transaction.Rollback();
            }
        }
    }

這就是典型ADO.NET事務處理代碼，其實和我們第一個例子中的T-SQL代碼是差不多的，通過ADO.NET中的SqlConnection.BeginTransaction()獲取到對底層ODBC中的數據庫事務的引用，其實這里還沒有真正的設計到.NET中的事務處理代碼，這里只是對數據庫管理系統的遠程調用，通過遠程處理的消息通訊進行事務處理遠程化。

事務信息顯示類，為了便於觀察事務的狀態信息。

 

public class DisplayTransactioninfo
    {
        public static void Display(System.Transactions.Transaction tr)
        {
            if (tr != null)
            {
                Console.WriteLine("Createtime:" + tr.TransactionInformation.CreationTime);
                Console.WriteLine("Status:" + tr.TransactionInformation.Status);
                Console.WriteLine("Local ID:" + tr.TransactionInformation.LocalIdentifier);
                Console.WriteLine("Distributed ID:" + tr.TransactionInformation.DistributedIdentifier);
                Console.WriteLine();
            }
        }

 

CommittableTransaction 事務處理

從這里開始我們將接觸到.NET中的事務處理，將了解到.NET中事務是怎樣影響到遠程數據庫管理系統的事務處理的。

其實事務處理是一個非常復雜的技術領域，需要考慮很多可逆的技術實現，我們只是簡單的了解原理和掌握基本的運用。

在我的 “ .NET簡談事務本質論 ”一文中說到了事務的傳遞原理，那么事務傳遞意味着什么。其實事務傳遞的大概意思是將事務的執行范圍通過網絡傳輸的方式進行擴大到其他的機器上，比如我們在.NET中執行一項關於事務性的操作，那么在這個操作里面我們包含了對數據庫的操作，這個時候對數據庫的一系列操作都應該是屬於事務范圍內的，當事務回滾時還應該將數據庫中的數據進行回滾才對。但是我們不可能總是顯示的執行ADO.NET中的BeginTransaction，對於本地事務處理也就是單一資源管理器來說這也可以接受，那么如果在事務范圍內涉及到多個資源管理器的操作，這就是分布式事務處理的范圍了。所以說事務處理需要跨越網絡傳輸形成無縫的面向服務的事務處理，數據庫管理系統即有可能扮演者事務管理器的角色也有可能扮演着資源管理器的角色。太多的理論知識我這里就不多扯了，我們還是來看代碼吧。

接着上面的實例，我們現在通過.NET中的事務處理來進行自動化的數據庫事務處理。讓數據庫能自動的感知到我們正在進行事務性的操作。

例 3 ：

我們利用Transaction類的子類CommittableTransaction可提交事務類來進行事務編程。

 

public class Test3
    {
        SqlConnection conn;
        CommittableTransaction committran = new CommittableTransaction();
        public Test3()
        {
            conn = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
            DisplayTransactioninfo.Display(committran);
        }
        public void Add3()
        {
            conn.Open();
            conn.EnlistTransaction(committran);//需要將本次的連接操作視為事務性的
            SqlCommand command = new SqlCommand();
            try
            {
                command.Connection = conn;
                command.CommandText = "insert into test2 values(111)";
                command.ExecuteNonQuery();
                command.CommandText = "insert into test values(222)";
                command.ExecuteNonQuery();
                committran.Commit();
            }
            catch (Exception err) { committran.Rollback(); //出現出錯執行回滾操作}
        }
    }

 

數據源連接對象代表着遠程數據庫資源，所以在執行操作之前我們需要將資源管理器添加到本地事務管理器中進行后期的投票、提交管理。

 

EnterpriseService(COM+) 自動化事務處理

在.NET2.0中有一個程序集不是太被人重視，System.EnterpriseServices.dll，這個程序集是.NET中為了使用早起的COM+技術的托管程序集，我們可以使用這個程序集來編寫一些我們以前所不能編寫的COM+應用程序。至於COM+應用程序的介紹網上一大堆，隨便搜搜就有好多資料了，我印象中有一篇是潘愛明潘老師寫的一個文章蠻好的，就是介紹COM+的所有特性。

我們繼續來事務處理，下面我們看看怎么借用System.EnterpriseServices.Transaction類來進行自動化的事務處理。

例 4 ：

 

[System.Runtime.InteropServices.ComVisible(true)]
//COM+是在COM的基礎上發展起來的，需要將.NET程序集中的類型公開為COM組件。
    [System.EnterpriseServices.Transaction(TransactionOption.Required)]//始終需要事務處理域
    public class Test2 : ServicedComponent
    {
        public Test2() { }
        SqlConnection conn = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        [AutoComplete(true)]
//如果在方法的執行體類沒有出現錯誤，那么將自動設置事務處理的結果
        public void Add2()
        {
            conn.Open();
            SqlCommand command = new SqlCommand();
            try
            {
                command.Connection = conn;
                command.CommandText = "insert into test2 values(111)";
                command.ExecuteNonQuery();
                command.CommandText = "insert into test values(222)";
                command.ExecuteNonQuery();
            }
            catch { System.EnterpriseServices.ContextUtil.SetAbort(); }
        }
    }

 

DependentTransaction 跨線程事務處理

我們在編寫高並發量程序時，都會用到多線程來進行處理，讓主線程能有時間來處理第一線的請求，然后將請求分發到各個子線程上進行后台的處理。我們來看一幅圖：

我們假設上面這幅圖是我們系統的一個內部結構，主線程主要的任務就是接受外來的請求，然后將具體的任務完成放到一個到兩個子線程中去完成，但是子線程與子線程之間沒有必然的關系，由於事務的上下文是不誇線程的，那么怎么將兩個或者更多的線程串在一個事務里。 [王清培版權所有，轉載請給出署名]

我們來看看依賴事務處理，看代碼：

例 5 ：

 

public class Test6
    {
        CommittableTransaction commit = new CommittableTransaction();
        SqlConnection conn1 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        public Test6()
        {
            conn1.Open();
            conn1.EnlistTransaction(commit);
        }
        public void Add6()
        {
            try
            {
                DisplayTransactioninfo.Display(commit);
                SqlCommand command = new SqlCommand("insert into test2 values(111)", conn1);
                command.ExecuteNonQuery();
                Thread thread = new Thread(Test6.CommitThread);
            thread.Start(commit.DependentClone(DependentCloneOption.BlockCommitUntilComplete));
                commit.Commit();
            }
            catch (Exception err) { commit.Rollback(); }
        }
        public static void CommitThread(object co)
        {
            DependentTransaction commit = co as DependentTransaction;
            SqlConnection conn2 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
            conn2.Open();
            conn2.EnlistTransaction(commit as DependentTransaction);
            DisplayTransactioninfo.Display(commit);
            SqlCommand command = new SqlCommand("insert into test values(111)", conn2);
            try
            {
                command.ExecuteNonQuery();
                commit.Complete();
            }
            catch (Exception err) { Console.WriteLine(err); commit.Rollback(); }
        }
    }

 

我們用一個子線程來執行另外的一個事務處理，由於是依賴事務處理，所以主事務處理完成后要等待子事務處理的結果。其實本例子已經是涉及到分布式事務處理的范圍了，當事務范圍內有一個以上的資源管理器時，本地事務管理器將自動提升為DTC管理器，下面我們來看看分布式事務處理。

 

DTC( Distributed Transaction Coordinator) 分布式事務處理

分布式事務在開發中經常是被用到，也必須被用到。必須同步數據、上下文更新等等。

按照使用方式的不同分布式事務的復雜程度也不同，基於本地事務的多資源管理器和基於SOA的面向服務的多資源管理器。

由於本地事務處理是基於本地事務管理器的，所以它不能管理分布式的事務，一旦當我們處理的事務范圍要進行擴大時並且是誇機器的訪問時，那么本地事務管理器將自動提升為分布式事務管理器也就是DTC(分布式事務協調器)。

例 6 ：

 

public class Test4
    {
        SqlConnection conn1 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        SqlConnection conn2 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        CommittableTransaction committran = new CommittableTransaction();
        public Test4()
        {
            DisplayTransactioninfo.Display(committran);
            conn1.Open();
            conn1.EnlistTransaction(committran);
            conn2.Open();
            conn2.EnlistTransaction(committran);
            DisplayTransactioninfo.Display(committran);
        }
        public void Add4()
        {
            try
            {
                SqlCommand command1 = new SqlCommand("insert into test2 values(111)", conn1);
                command1.ExecuteNonQuery();
                SqlCommand command2 = new SqlCommand("insert into test values(222)", conn2);
                command2.ExecuteNonQuery();
            }
            catch (Exception err) { Console.WriteLine(err); committran.Rollback(); }
        }
    }

一旦我們開啟分布式事務處理就能在我們的電腦上的DTC管理器上看見到。

 

但是要記得檢查你的DTC服務是否開啟了。

 

基於 WCF 框架的分布式事務處理

其實基於WCF框架進行分布式事務開發真的很輕松，它能很好的感知到當前上下文是不是事務域，並能很好的將事務序列化到服務這邊來。但是設計一個高性能的分布式事務處理框架並非易事，需要很長時間的積累和實踐。我們來看一下WCF是如果進行分布式事務處理的。

例 7 ：

 

//服務契約(ServiceContract):
[ServiceContract(SessionMode = SessionMode.Required)]
    public interface IDistributedTransaction
    {
        [TransactionFlow(TransactionFlowOption.Allowed)]
        [OperationContract]
        void Add();
    }

 

//服務類1：
public class DistributedTransactionService1 : IDistributedTransaction
    {
        SqlConnection conn1 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        #region IDistributedTransaction
        [OperationBehavior(TransactionAutoComplete = true, TransactionScopeRequired = true)]
        public void Add()
        {
            conn1.Open();
            SqlCommand command = new SqlCommand("insert into test2 values(111)", conn1);
            command.ExecuteNonQuery();
            DataBaseOperation.DisplayTransactioninfo.Display(System.Transactions.Transaction.Current);
        }
        #endregion
    }

 

//服務類2：
public class DistributedTransactionService2 : IDistributedTransaction
    {
        SqlConnection conn2 = new SqlConnection("data source=.;Initial Catalog=DataMedicine;Integrated Security=SSPI");
        #region IDistributedTransaction
        [OperationBehavior(TransactionAutoComplete = true, TransactionScopeRequired = true)]
        public void Add()
        {
            conn2.Open();
            SqlCommand command = new SqlCommand("insert into test values(222)", conn2);
            try
            {
                DataBaseOperation.DisplayTransactioninfo.Display(System.Transactions.Transaction.Current);
                command.ExecuteNonQuery();
            }
            catch (Exception err) { throw err; }
        }

服務配置：

<service name="ServerConsole.Transaction.DistributedTransactionService1" behaviorConfiguration="metadatabehaviors">
        <host>
          <baseAddresses>
            <add baseAddress="http://localhost:8027"/>
            <add baseAddress="net.tcp://localhost:8028"/>
          </baseAddresses>
        </host>
        <endpoint address="DistributedTransactionService1" binding="netTcpBinding" bindingConfiguration="tranbinding"
                   contract="ServerConsole.Transaction.IDistributedTransaction"></endpoint>
      </service>
      <service name="ServerConsole.Transaction.DistributedTransactionService2" behaviorConfiguration="metadatabehaviors">
        <host>
          <baseAddresses>
            <add baseAddress="http://localhost:8029"/>
            <add baseAddress="net.tcp://localhost:8030"/>
          </baseAddresses>
        </host>
        <endpoint address="DistributedTransactionService2" binding="netTcpBinding" bindingConfiguration="tranbinding"
                contract="ServerConsole.Transaction.IDistributedTransaction"></endpoint>
      </service>

Binding配置：

<bindings>
      <netTcpBinding>
        <binding name="tranbinding" transactionFlow="true" transactionProtocol="WSAtomicTransactionOctober2004">
          <reliableSession enabled="true" ordered="true"/>
        </binding>
      </netTcpBinding>
    </bindings>

我們需要打開Binding的事務流傳遞。

客戶端代碼：

客戶端：
DistributedTransactionClient.DistributedTransactionClient tranclient = new DistributedTransactionClient.DistributedTransactionClient();
            DistributedTransaction2Client.DistributedTransactionClient tranclient2 = new DistributedTransaction2Client.DistributedTransactionClient();
            using (TransactionScope transcope = new TransactionScope())
            {
                try
                {
                    Transaction.Current.TransactionCompleted += new TransactionCompletedEventHandler(Current_TransactionCompleted);
                    tranclient.Add();
                    tranclient2.Add();
                    transcope.Complete();
                }
                catch (Exception err) { Transaction.Current.Rollback(); }
            }

static void Current_TransactionCompleted(object sender, TransactionEventArgs e)
        {
            if (e.Transaction.TransactionInformation.Status ==
                System.Transactions.TransactionStatus.Committed)
                Console.WriteLine(e.Transaction.TransactionInformation.DistributedIdentifier);
        }

 

客戶端使用TransactionScope類來進行環境事務的設置，這樣就很方便知道事務的執行范圍，在TransactionScope里面我們可以通過Transaction.Current獲取到當前上下文的事務對象，由於事務對象是存儲在線程獨立存儲區里的，所以跨線程訪問是沒用的，通過依賴事務進行傳遞。

 

文章到這里就講完了，從本地事務、多資源管理器分布式事務、SOA結構的分布式事務，我們都能進行基本的掌握。上面的例子都是經過嚴格測試的。