.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结构的分布式事务，我们都能进行基本的掌握。上面的例子都是经过严格测试的。