我有一个应用程序,是实现数据ETL同步的,即把数据从一个db里抽取出来,经过处理后,存储到另一个db里。 O/RM采用的是EF db First。
随着项目程序的开发,EF的不足越来越不足。
● 根据EDM生成的类,没有继承关系,影响程序设计实现
我是直接根据edmx文件生成的类, 每个数据表对应一个class。 问题来了,这些类各自独立,没有面向对象里的封装和继承。
我的设计:各个数据同步类,抽象出来了一个基类,封装了对外接口方法、日志记录、告警、还有一些共有的处理方法等功能。基类是一个泛型类,T是代表的是POCO,由于EF生成的POCO并没有继承相同的基类,无奈我的这个基类只好这样定义:
public abstract class BizOrderETLBase<T> where T : class { private DateTime _timeFrom; private DateTime _timeTo; protected DateTime _TimeFrom { get { return _timeFrom; } } protected DateTime _TimeTo { get { return _timeTo; } } public BizOrderETLBase(DateTime timeFrom, DateTime timeTo) { _timeFrom = timeFrom; _timeTo = timeTo; } public BizOrderETLBase() : this(DateTime.Today.AddDays(-1), DateTime.Today) { } /// <summary> /// 执行ETL:Extract-Transform-Load /// </summary> /// <returns></returns> public int DoETL() { LogHelper.Write(">>>>>开始同步业务订单{0} {1}~{2}...", this.GetType().Name,_timeFrom,_timeTo); try { // 数据抽取和转换 var list = ExtractAndTransform(); if (list != null && list.Any()) { // 数据装载到OLAP库 int i = Load(list); LogHelper.Write("持久化共影响{0}条记录", i); return i; } } catch (Exception ex) { LogHelper.Write("同步业务订单{0}出现异常:\r\n{1}", this.GetType().Name, ex); } return 0; } /// <summary> /// ETL之ET (Extract 和 Transform) /// </summary> /// <returns></returns> protected abstract List<T> ExtractAndTransform(); /// <summary> /// ETL之L (Load) /// </summary> /// <param name="datOrderList"></param> /// <returns></returns> private int Load(List<T> datOrderList) { // 持久化数据 var olapDal = new EFDbContext<T>(); int i = 0; // 先删除之前生成的统计数据(如果有) Expression<Func<T, bool>> where = GetDeleteExpression(); LogHelper.Write("持久化删除_开始"); i = olapDal.DeleteByWhere(where); LogHelper.Write("持久化删除_结束"); // 保存本次的统计数据 return olapDal.Add(datOrderList.ToArray()); } /// <summary> /// 得到删除数据的条件 /// </summary> /// <returns></returns> protected abstract Expression<Func<T, bool>> GetDeleteExpression(); }
Load方法的功能主要是对已处理的数据进行保存,保存之前先删除现有数据。 删除的条件和数据抽取的条件一样,都是按一个名为ModifiedTime的时间戳字段判断的。 因为T被声明为了class类型,只好加了一个得到删除数据的条件的抽象方法GetDeleteExpression
各个继承类里都有几乎相同的代码,造成代码重复:
protected override Expression<Func<BizDatOrders, bool>> GetDeleteExpression() { Expression<Func<BizDatOrders, bool>> where = p => p.OrderModifiedTime >= _TimeFrom && p.OrderModifiedTime <= _TimeTo; return where; }
●EF无法实现事务隔离
由于数据源db的IO大,我这个ETL程序在从数据源获取数据时,经常出现死锁而被当成牺牲品。为了改进,不得不引入事务隔离级别,实现read uncommitted, 用写sql的方式很容易实现,在各表名后加nolock就可以了。 而EF没有现成的事务隔离,我使用了分布式事务TransactionScope。 问题来了,在TransactionScope里有多个不同连接的DbContext时,出现异常:与基础事务管理器的通信失败。
代码如下:
public class NoLockHelper { public delegate int EFdelegate(); public int NoLockInvokeDB(EFdelegate d) { var transactionOptions = new System.Transactions.TransactionOptions(); transactionOptions.IsolationLevel = System.Transactions.IsolationLevel.ReadUncommitted; using (var transactionScope = new TransactionScope(System.Transactions.TransactionScopeOption.Required, transactionOptions)) { int i = d.Invoke(); transactionScope.Complete(); return i; } } } public class BizClass { public void Test() { new NoLockHelper().NoLockInvokeDB(Target); } public int Target() { TravelPPEntities _travelPPContext = new TravelPPEntities(); var _airOrderRepository = _travelPPContext.Set<T_Business_AirOrders>(); var list = _airOrderRepository.Count(); Thread.Sleep(3 * 1000); Console.WriteLine(list); _travelPPContext.Dispose(); OLAPEntities db = new OLAPEntities(); var lll = db.Set<BaseAirport>().ToList(); // 执行这句会报异常 return list; } }
异常信息:
System.Data.EntityException: 基础提供程序在 Open 上失败。 ---> System.Transactions.TransactionManagerCommunicationException: 与基础事务管理器的通信失败。 ---> System.Runtime.InteropServices.COMException: 事务管理器不可用。 (异常来自 HRESULT:0x8004D01B)
在 System.Transactions.Oletx.IDtcProxyShimFactory.ConnectToProxy(String nodeName, Guid resourceManagerIdentifier, IntPtr managedIdentifier, Boolean& nodeNameMatches, UInt32& whereaboutsSize, CoTaskMemHandle& whereaboutsBuffer, IResourceManagerShim& resourceManagerShim)
在 System.Transactions.Oletx.DtcTransactionManager.Initialize()
--- 内部异常堆栈跟踪的结尾 ---
在 System.Transactions.Oletx.OletxTransactionManager.ProxyException(COMException comException)
在 System.Transactions.Oletx.DtcTransactionManager.Initialize()
在 System.Transactions.Oletx.DtcTransactionManager.get_ProxyShimFactory()
在 System.Transactions.TransactionInterop.GetOletxTransactionFromTransmitterPropigationToken(Byte[] propagationToken)
在 System.Transactions.TransactionStatePSPEOperation.PSPEPromote(InternalTransaction tx)
在 System.Transactions.TransactionStateDelegatedBase.EnterState(InternalTransaction tx)
在 System.Transactions.EnlistableStates.Promote(InternalTransaction tx)
在 System.Transactions.Transaction.Promote()
在 System.Transactions.TransactionInterop.ConvertToOletxTransaction(Transaction transaction)
在 System.Transactions.TransactionInterop.GetExportCookie(Transaction transaction, Byte[] whereabouts)
在 System.Data.SqlClient.SqlInternalConnection.GetTransactionCookie(Transaction transaction, Byte[] whereAbouts)
在 System.Data.SqlClient.SqlInternalConnection.EnlistNonNull(Transaction tx)
在 System.Data.SqlClient.SqlInternalConnection.Enlist(Transaction tx)
在 System.Data.SqlClient.SqlInternalConnectionTds.Activate(Transaction transaction)
在 System.Data.ProviderBase.DbConnectionInternal.ActivateConnection(Transaction transaction)
在 System.Data.ProviderBase.DbConnectionPool.PrepareConnection(DbConnection owningObject, DbConnectionInternal obj, Transaction transaction)
在 System.Data.ProviderBase.DbConnectionPool.TryGetConnection(DbConnection owningObject, UInt32 waitForMultipleObjectsTimeout, Boolean allowCreate, Boolean onlyOneCheckConnection, DbConnectionOptions userOptions, DbConnectionInternal& connection)
在 System.Data.ProviderBase.DbConnectionPool.TryGetConnection(DbConnection owningObject, TaskCompletionSource`1 retry, DbConnectionOptions userOptions, DbConnectionInternal& connection)
在 System.Data.ProviderBase.DbConnectionFactory.TryGetConnection(DbConnection owningConnection, TaskCompletionSource`1 retry, DbConnectionOptions userOptions, DbConnectionInternal oldConnection, DbConnectionInternal& connection)
在 System.Data.ProviderBase.DbConnectionInternal.TryOpenConnectionInternal(DbConnection outerConnection, DbConnectionFactory connectionFactory, TaskCompletionSource`1 retry, DbConnectionOptions userOptions)
在 System.Data.ProviderBase.DbConnectionClosed.TryOpenConnection(DbConnection outerConnection, DbConnectionFactory connectionFactory, TaskCompletionSource`1 retry, DbConnectionOptions userOptions)
在 System.Data.SqlClient.SqlConnection.TryOpenInner(TaskCompletionSource`1 retry)
在 System.Data.SqlClient.SqlConnection.TryOpen(TaskCompletionSource`1 retry)
在 System.Data.SqlClient.SqlConnection.Open()
在 System.Data.EntityClient.EntityConnection.OpenStoreConnectionIf(Boolean openCondition, DbConnection storeConnectionToOpen, DbConnection originalConnection, String exceptionCode, String attemptedOperation, Boolean& closeStoreConnectionOnFailure)
--- 内部异常堆栈跟踪的结尾 ---
在 System.Data.EntityClient.EntityConnection.OpenStoreConnectionIf(Boolean openCondition, DbConnection storeConnectionToOpen, DbConnection originalConnection, String exceptionCode, String attemptedOperation, Boolean& closeStoreConnectionOnFailure)
在 System.Data.EntityClient.EntityConnection.Open()
在 System.Data.Objects.ObjectContext.EnsureConnection()
在 System.Data.Objects.ObjectQuery`1.GetResults(Nullable`1 forMergeOption)
在 System.Data.Objects.ObjectQuery`1.System.Collections.Generic.IEnumerable<T>.GetEnumerator()
在 System.Data.Entity.Internal.Linq.InternalQuery`1.GetEnumerator()
在 System.Data.Entity.Internal.Linq.InternalSet`1.GetEnumerator()
在 System.Data.Entity.Infrastructure.DbQuery`1.System.Collections.Generic.IEnumerable<TResult>.GetEnumerator()
在 System.Collections.Generic.List`1..ctor(IEnumerable`1 collection)
在 System.Linq.Enumerable.ToList[TSource](IEnumerable`1 source)
在 EntOlap.ETL.EF.EFDbContext`1.GetList() 位置 d:\SourceProject\OLAP\trunk\EntOlap\EntOlap.ETL\EntOlap.ETL.EF\EFDbContext.cs:行号 347