前言：
由于系统升级，新开发的系统对数据验证，及数据关联做了很多优化，现需要将原历史版本的数据迁移到新系统中；原数据库大约有 1千多万数据，大约 50个表。
历史数据库命名为：A。 新系统库暂命名为 ：B；
使用 .net 4.5 控制台程序 + EF + MSSQL 数据库，由于有业务逻辑及时序处理，故只能按时序从单表一条条的写入到新库中；
化化过程：
1、EF 如果使用多线程会出现 Sql 连接超过，或是连接不上数据库；
2、EF 优化连接 自定义 SqlConnection，并传到入 多线程中，解决连接不上数据库的问题减少数据库连接数，但由于 EF 在 SaveChangesAsync的时候做了事务提交，但事务是不支持并行操作，故会出现异常；
3、EF 优化事务，关闭EF默认事务 DbContextConfiguration.EnsureTransactionsForFunctionsAndCommands = false; 这里有个坑 关闭事务对 SaveChangesAsync 无效，问题依然存在；
4、找了很多资料总算找到可以通过 ExecuteSqlCommandAsync 执行 Sql 语句，可以关闭事务；
5、优化成执行Sql 语句：await db.Database.ExecuteSqlCommandAsync(TransactionalBehavior.DoNotEnsureTransaction, sql, SqlParameters[]);
经过以上优化处理后，就开始写代码：
一、关键的异步锁程序：
/// <summary> /// 提供异步锁 /// </summary> class AsyncRoot : IDisposable { /// <summary> /// 信号量 /// </summary> private readonly SemaphoreSlim semaphoreSlim; /// <summary> /// 异步锁 /// </summary> public AsyncRoot() : this(1) { } /// <summary> /// 异步锁 /// </summary> /// <param name="concurrent">允许并行的线程数</param> public AsyncRoot(int concurrent) { this.semaphoreSlim = new SemaphoreSlim(concurrent, concurrent); } /// <summary> /// 锁住代码块 /// using( asyncRoot.Lock() ){ } /// </summary> /// <returns></returns> public IDisposable Lock() { this.semaphoreSlim.Wait(); return new UnLocker(this.semaphoreSlim); } /// <summary> /// 锁住代码块 /// using( await asyncRoot.LockAsync() ){ } /// </summary> /// <returns></returns> public async Task<IDisposable> LockAsync() { await this.semaphoreSlim.WaitAsync().ConfigureAwait(false); return new UnLocker(this.semaphoreSlim); } /// <summary> /// 释放资源 /// </summary> public void Dispose() { this.semaphoreSlim.Dispose(); } /// <summary> /// 提供解锁 /// </summary> class UnLocker : IDisposable { /// <summary> /// 信号量 /// </summary> private readonly SemaphoreSlim semaphoreSlim; /// <summary> /// 解锁 /// </summary> /// <param name="semaphoreSlim">信号量</param> public UnLocker(SemaphoreSlim semaphoreSlim) { this.semaphoreSlim = semaphoreSlim; } /// <summary> /// 释放锁 /// </summary> public void Dispose() { this.semaphoreSlim.Release(); } } }
二、对数据插入到数据库：
逻辑分析：对传入的 数据集合，拆分为单个实体操作任务，每个任务使用同一个连接，独立的数据库上下文，对实体反射为 Sql 语句(其中增加主键，表名、字段名、值的判断验证)，
然后通过 ExecuteSqlCommandAsync 不使用事务的方式执行 Sql 语句；具体代码见下：
//表示最大线程数
private readonly AsyncRoot root = new AsyncRoot(50);
/// <summary> /// 多线程工作 /// </summary> public class Workers { /// <summary> /// 多线程锁 /// </summary> private readonly AsyncRoot root = new AsyncRoot(50); /// <summary> /// 执行对象操作 /// </summary> /// <param name="datas"></param> /// <returns></returns> public async Task RunAsync<T>(IEnumerable<T> datas) where T : class { //创建 Sql 连接 var connection = new SqlConnection(System.Configuration.ConfigurationManager.ConnectionStrings["SqlDb"].ConnectionString); await connection.OpenAsync(); var tasks = datas.Select(item => SaveToDbAsync(item, connection)); await Task.WhenAll(tasks); } /// <summary> /// 单条记录保存到数据库 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="data"></param> /// <param name="connection"></param> /// <returns></returns> private async Task SaveToDbAsync<T>(T data, DbConnection connection) where T : class { using (await root.LockAsync()) { using (var db = new SqlDb(connection)) { try { var dbset = db.Set<T>(); var tType = typeof(T); var tableName = tType.Name; //获取 TableAttribute 数据库中的表名 var tableAtt = Attribute.GetCustomAttribute(tType, typeof(TableAttribute)) as TableAttribute; if (tableAtt != null) { tableName = tableAtt.Name; } var sbSql = new StringBuilder(); sbSql.AppendLine("insert into " + tableName + " ("); var plist = new List<string>(); var fieldParameters = new List<SqlParameter>(); var keyFiled = "ID"; foreach (var p in typeof(T).GetProperties()) { var pName = p.Name.ToUpper(); //获取 ColumnAttribute 数据库中的列名 var pAtt = Attribute.GetCustomAttribute(p, typeof(ColumnAttribute)) as ColumnAttribute; if (pAtt != null) { pName = pAtt.Name.ToUpper(); } var keyAtt = Attribute.GetCustomAttribute(p, typeof(KeyAttribute)) as KeyAttribute; if (keyAtt != null || p.Name.Equals("ID", StringComparison.OrdinalIgnoreCase)) { keyFiled = pName; } var fieldParameter = "@" + pName; //过滤不插入数据库中的字段 var mapAtt = Attribute.GetCustomAttribute(p, typeof(NotMappedAttribute)); if (mapAtt == null) { var value = p.GetValue(data, null); //如果属性值为 Null，不插入数据库 if (value != null) { plist.Add(fieldParameter); fieldParameters.Add(new SqlParameter(fieldParameter, value)); } } } sbSql.Append(string.Join(",", plist.Select(item => item.Replace("@", "")))); sbSql.Append(")values("); sbSql.Append(string.Join(",", plist)); sbSql.Append(")"); //判断主键是否已经存在,存在就不插入数据 var ifSql = "if not exists(select 1 from [" + tableName + "] where " + keyFiled + " = @" + keyFiled + ")"; var sql = ifSql + sbSql.ToString(); await db.Database.ExecuteSqlCommandAsync(TransactionalBehavior.DoNotEnsureTransaction, sql, fieldParameters.ToArray()); } catch (Exception ex) { Console.WriteLine(ex); } } } } }
/// <summary> /// Sql数据库 /// </summary> public class SqlDb : DbContext { /// <summary> /// 自定义连接 /// </summary> /// <param name="connection">数据库连接</param> public SqlDb(DbConnection connection) : base(connection, false) { if (connection.State != System.Data.ConnectionState.Open) { connection.Open(); } this.Database.CommandTimeout = 60 * 1000; this.Configuration.UseDatabaseNullSemantics = true; this.Configuration.EnsureTransactionsForFunctionsAndCommands = false; this.Configuration.ValidateOnSaveEnabled = false; } }
三、注意事项：
1、如果字段为 geography (地理位置) 类型，会出现异常，希望在使用的时候注意一下；
2、由于集合为同一个对象，故在每次反射的对象几乎都是重复操作，可以根据实际情况增加缓存；
其它：
多线程并行操作小实例源码：https://github.com/intotf/netExample/tree/master/Tool/MultiTaskAsync