C# 封裝SqlBulkCopy，讓批量插入更方便
阿新 • • 發佈：2018-09-22
creat 解決 lln index initial 利用反射 eve exe pub 
關於 SqlServer 批量插入的方式，前段時間也有大神給出了好幾種批量插入的方式及對比測試（http://www.cnblogs.com/jiekzou/p/6145550.html），估計大家也都明白，最佳的方式就是用 SqlBulkCopy。我對 SqlBulkCopy 封裝成了一個 Helper 方法，使得批量插入更加方便，先看看封裝後的方法定義：
public static class SqlConnectionExtension { /// <summary> /// 使用 SqlBulkCopy 向 destinationTableName 表插入數據 /// </summary>/// <typeparam name="TModel">必須擁有與目標表所有字段對應屬性</typeparam> /// <param name="conn"></param> /// <param name="modelList">要插入的數據</param> /// <param name="batchSize">SqlBulkCopy.BatchSize</param> /// <param name="destinationTableName">如果為 null，則使用 TModel 名稱作為 destinationTableName</param> /// <param name="bulkCopyTimeout">SqlBulkCopy.BulkCopyTimeout</param> /// <param name="externalTransaction">要使用的事務</param> public static void BulkCopy<TModel>(this SqlConnection conn, List<TModel> modelList, int batchSize, string destinationTableName = null, int? bulkCopyTimeout = null, SqlTransaction externalTransaction = null); }
上面都有詳細解釋，相信大家一看就會明白，接下來演示下用法及效果：
先創建一個測試的 Users 表：
CREATE TABLE [dbo].[Users]( [Id] [uniqueidentifier] NOT NULL, [Name] [nvarchar](100) NULL, [Gender] [int] NULL, [Age] [int] NULL, [CityId] [int] NULL, [OpTime] [datetime] NULL, CONSTRAINT [PK_Users] PRIMARY KEY CLUSTERED([Id] ASC)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY] ) ON [PRIMARY]
然後定義一個與表映射的 Model，記住，由於 SqlBulkCopy 的特性，定義的 Model 必須擁有與表所有的字段對應的屬性：
public enum Gender { Man = 1, Woman } public class User { public Guid Id { get; set; } public string Name { get; set; } public Gender? Gender { get; set; } public int? Age { get; set; } public int? CityId { get; set; } public DateTime? OpTime { get; set; } }
制造些數據，然後就可以直接插入了：
List<User> usersToInsert = new List<User>(); usersToInsert.Add(new User() { Id = Guid.NewGuid(), Name = "so1", Gender = Gender.Man, Age = 18, CityId = 1, OpTime = DateTime.Now }); usersToInsert.Add(new User() { Id = Guid.NewGuid(), Name = "so2", Gender = Gender.Man, Age = 19, CityId = 2, OpTime = DateTime.Now }); usersToInsert.Add(new User() { Id = Guid.NewGuid(), Name = "so3", Gender = Gender.Man, Age = 20, CityId = 3, OpTime = DateTime.Now }); usersToInsert.Add(new User() { Id = Guid.NewGuid(), Name = "so4", Gender = Gender.Man, Age = 21, CityId = 4, OpTime = DateTime.Now }); using (SqlConnection conn = new SqlConnection("Data Source = .;Initial Catalog = Chloe;Integrated Security = SSPI;")) { conn.BulkCopy(usersToInsert, 20000, "Users"); }
執行插入後表數據：
很方便吧，定義好 Model，調用 BulkCopy 方法就能插入了。這個方法主要解決了兩個問題：1.免去手動構造 DataTable 和向 DataTable 填充數據，要知道，SqlBulkCopy 要求 DataTable 的列必須和表列順序一致，如果手動構造 DataTable 的話會使代碼很難維護；2.不用親自 new 出 SqlBulkCopy 對象以及手動給 SqlBulkCopy 對象設置各種值，如 DestinationTableName、BulkCopyTimeout、BatchSize 等，用封裝的方法，直接傳相應的值就好了。接下來貼幹貨，簡單介紹下實現。
先了解 SqlBulkCopy 的定義（部分）：
public sealed class SqlBulkCopy : IDisposable { public SqlBulkCopy(SqlConnection connection); public SqlBulkCopy(string connectionString); public SqlBulkCopy(string connectionString, SqlBulkCopyOptions copyOptions); public SqlBulkCopy(SqlConnection connection, SqlBulkCopyOptions copyOptions, SqlTransaction externalTransaction); public int BatchSize { get; set; } public int BulkCopyTimeout { get; set; } public SqlBulkCopyColumnMappingCollection ColumnMappings { get; } public string DestinationTableName { get; set; } public bool EnableStreaming { get; set; } public int NotifyAfter { get; set; } public event SqlRowsCopiedEventHandler SqlRowsCopied; public void Close(); public void WriteToServer(DataRow[] rows); public void WriteToServer(DataTable table); public void WriteToServer(IDataReader reader); public void WriteToServer(DataTable table, DataRowState rowState); }
我們只需關註 WriteToServer 方法。因為我們的數據源不是數據庫或excel，所以我們直接不考慮 WriteToServer(IDataReader reader)。WriteToServer(DataRow[] rows) 直接無視，不多解釋，所以我們只需考慮用 WriteToServer(DataTable table) 就行了。開幹！
一、構造一個結構嚴謹的 DataTable。
由於 SqlBulkCopy 要求 DataTable 的列必須和表列順序一致，並且不能多也不能少，所以，我們首先要創建一個和目標表字段順序一致的 DataTable，先查出目標表的結構：
static List<SysColumn> GetTableColumns(SqlConnection sourceConn, string tableName) { string sql = string.Format("select * from syscolumns inner join sysobjects on syscolumns.id=sysobjects.id where sysobjects.xtype=‘U‘ and sysobjects.name=‘{0}‘ order by syscolumns.colid asc", tableName); List<SysColumn> columns = new List<SysColumn>(); using (SqlConnection conn = (SqlConnection)((ICloneable)sourceConn).Clone()) { conn.Open(); using (var reader = conn.ExecuteReader(sql)) { while (reader.Read()) { SysColumn column = new SysColumn(); column.Name = reader.GetDbValue("name"); column.ColOrder = reader.GetDbValue("colorder"); columns.Add(column); } } conn.Close(); } return columns; }
得到基本的表結構 List<SysColumn>，再創建“嚴格”的 DataTable 對象：
DataTable dt = new DataTable(); Type modelType = typeof(TModel); List<SysColumn> columns = GetTableColumns(conn, tableName); List<PropertyInfo> mappingProps = new List<PropertyInfo>(); var props = modelType.GetProperties(); for (int i = 0; i < columns.Count; i++) { var column = columns[i]; PropertyInfo mappingProp = props.Where(a => a.Name == column.Name).FirstOrDefault(); if (mappingProp == null) throw new Exception(string.Format("model 類型 ‘{0}‘未定義與表 ‘{1}‘ 列名為 ‘{2}‘ 映射的屬性", modelType.FullName, tableName, column.Name)); mappingProps.Add(mappingProp); Type dataType = GetUnderlyingType(mappingProp.PropertyType); if (dataType.IsEnum) dataType = typeof(int); dt.Columns.Add(new DataColumn(column.Name, dataType)); }
註意，構造 DataColumn 時，要給 Column 設置 DataType，及數據類型。因為如果不指定數據類型，默認是 string 類型，那樣會導致將數據發送至數據庫時會引起數據轉換，會有些許無謂的性能損耗，同時，如果不指定數據類型，導入一些數據類型時可能會失敗，比如模型屬性是 Guid 類型，導入時會出現類型轉換失敗異常。
二、利用反射，獲取屬性值，構造一行一行的 DataRow，填充 DataTable：
foreach (var model in modelList) { DataRow dr = dt.NewRow(); for (int i = 0; i < mappingProps.Count; i++) { PropertyInfo prop = mappingProps[i]; object value = prop.GetValue(model); if (GetUnderlyingType(prop.PropertyType).IsEnum) { if (value != null) value = (int)value; } dr[i] = value ?? DBNull.Value; } dt.Rows.Add(dr); }
三、一個完整包含數據的 DataTable 對象就創建好了，我們就可以使用 SqlBulkCopy 插入數據了：
public static void BulkCopy<TModel>(this SqlConnection conn, List<TModel> modelList, int batchSize, string destinationTableName = null, int? bulkCopyTimeout = null, SqlTransaction externalTransaction = null) { bool shouldCloseConnection = false; if (string.IsNullOrEmpty(destinationTableName)) destinationTableName = typeof(TModel).Name; DataTable dtToWrite = ToSqlBulkCopyDataTable(modelList, conn, destinationTableName); SqlBulkCopy sbc = null; try { if (externalTransaction != null) sbc = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, externalTransaction); else sbc = new SqlBulkCopy(conn); using (sbc) { sbc.BatchSize = batchSize; sbc.DestinationTableName = destinationTableName; if (bulkCopyTimeout != null) sbc.BulkCopyTimeout = bulkCopyTimeout.Value; if (conn.State != ConnectionState.Open) { shouldCloseConnection = true; conn.Open(); } sbc.WriteToServer(dtToWrite); } } finally { if (shouldCloseConnection && conn.State == ConnectionState.Open) conn.Close(); } }
完事，一個批量插入的 Helper 方法就這麽產生了，最終的完整實現如下：
public static class SqlConnectionExtension { /// <summary> /// 使用 SqlBulkCopy 向 destinationTableName 表插入數據 /// </summary> /// <typeparam name="TModel">必須擁有與目標表所有字段對應屬性</typeparam> /// <param name="conn"></param> /// <param name="modelList">要插入的數據</param> /// <param name="batchSize">SqlBulkCopy.BatchSize</param> /// <param name="destinationTableName">如果為 null，則使用 TModel 名稱作為 destinationTableName</param> /// <param name="bulkCopyTimeout">SqlBulkCopy.BulkCopyTimeout</param> /// <param name="externalTransaction">要使用的事務</param> public static void BulkCopy<TModel>(this SqlConnection conn, List<TModel> modelList, int batchSize, string destinationTableName = null, int? bulkCopyTimeout = null, SqlTransaction externalTransaction = null) { bool shouldCloseConnection = false; if (string.IsNullOrEmpty(destinationTableName)) destinationTableName = typeof(TModel).Name; DataTable dtToWrite = ToSqlBulkCopyDataTable(modelList, conn, destinationTableName); SqlBulkCopy sbc = null; try { if (externalTransaction != null) sbc = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, externalTransaction); else sbc = new SqlBulkCopy(conn); using (sbc) { sbc.BatchSize = batchSize; sbc.DestinationTableName = destinationTableName; if (bulkCopyTimeout != null) sbc.BulkCopyTimeout = bulkCopyTimeout.Value; if (conn.State != ConnectionState.Open) { shouldCloseConnection = true; conn.Open(); } sbc.WriteToServer(dtToWrite); } } finally { if (shouldCloseConnection && conn.State == ConnectionState.Open) conn.Close(); } } public static DataTable ToSqlBulkCopyDataTable<TModel>(List<TModel> modelList, SqlConnection conn, string tableName) { DataTable dt = new DataTable(); Type modelType = typeof(TModel); List<SysColumn> columns = GetTableColumns(conn, tableName); List<PropertyInfo> mappingProps = new List<PropertyInfo>(); var props = modelType.GetProperties(); for (int i = 0; i < columns.Count; i++) { var column = columns[i]; PropertyInfo mappingProp = props.Where(a => a.Name == column.Name).FirstOrDefault(); if (mappingProp == null) throw new Exception(string.Format("model 類型 ‘{0}‘未定義與表 ‘{1}‘ 列名為 ‘{2}‘ 映射的屬性", modelType.FullName, tableName, column.Name)); mappingProps.Add(mappingProp); Type dataType = GetUnderlyingType(mappingProp.PropertyType); if (dataType.IsEnum) dataType = typeof(int); dt.Columns.Add(new DataColumn(column.Name, dataType)); } foreach (var model in modelList) { DataRow dr = dt.NewRow(); for (int i = 0; i < mappingProps.Count; i++) { PropertyInfo prop = mappingProps[i]; object value = prop.GetValue(model); if (GetUnderlyingType(prop.PropertyType).IsEnum) { if (value != null) value = (int)value; } dr[i] = value ?? DBNull.Value; } dt.Rows.Add(dr); } return dt; } static List<SysColumn> GetTableColumns(SqlConnection sourceConn, string tableName) { string sql = string.Format("select * from syscolumns inner join sysobjects on syscolumns.id=sysobjects.id where sysobjects.xtype=‘U‘ and sysobjects.name=‘{0}‘ order by syscolumns.colid asc", tableName); List<SysColumn> columns = new List<SysColumn>(); using (SqlConnection conn = (SqlConnection)((ICloneable)sourceConn).Clone()) { conn.Open(); using (var reader = conn.ExecuteReader(sql)) { while (reader.Read()) { SysColumn column = new SysColumn(); column.Name = reader.GetDbValue("name"); column.ColOrder = reader.GetDbValue("colorder"); columns.Add(column); } } conn.Close(); } return columns; } static Type GetUnderlyingType(Type type) { Type unType = Nullable.GetUnderlyingType(type); ; if (unType == null) unType = type; return unType; } class SysColumn { public string Name { get; set; } public int ColOrder { get; set; } } }完整代碼
代碼不多，僅僅150行，大家可以直接拷走拿去用。其中用了反射，估計吃瓜群眾可能不淡定了~哈哈，如果你真有大數據插入需求，這點反射消耗相對大數據插入簡直九牛一毛，微乎其微，放心好了。
轉自：https://www.cnblogs.com/so9527/p/6193154.html
C# 封裝SqlBulkCopy，讓批量插入更方便