一、Sql Server插入方案介紹

關于 SqlServer 批量插入的方式，有三種比較常用的插入方式，Insert、BatchInsert、SqlBulkCopy,下面我們對比以下三種方案的速度

1.普通的`Insert`插入方法

public static void Insert(IEnumerable<Person> persons)
{
    using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;"))
    {
        con.Open();
        foreach (var person in persons)
        {
            using (var com = new SqlCommand(
                "INSERT INTO dbo.Person(Id,Name,Age,CreateTime,Sex)VALUES(@Id,@Name,@Age,@CreateTime,@Sex)",
                con))
            {
                com.Parameters.AddRange(new[]
                {
                    new SqlParameter("@Id", SqlDbType.BigInt) {Value = https://www.cnblogs.com/liuzhenliang/p/person.Id},
                    new SqlParameter("@Name", SqlDbType.VarChar, 64) {Value = https://www.cnblogs.com/liuzhenliang/p/person.Name},
                    new SqlParameter("@Age", SqlDbType.Int) {Value = https://www.cnblogs.com/liuzhenliang/p/person.Age},
                    new SqlParameter("@CreateTime", SqlDbType.DateTime)
                        {Value = https://www.cnblogs.com/liuzhenliang/p/person.CreateTime ?? (object) DBNull.Value},
                    new SqlParameter("@Sex", SqlDbType.Int) {Value = https://www.cnblogs.com/liuzhenliang/p/(int)person.Sex},
                });
                com.ExecuteNonQuery();
            }
        }
    }
}

2.拼接`BatchInsert`插入陳述句

public static void BatchInsert(Person[] persons)
{
    using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;"))
    {
        con.Open();
        var pageCount = (persons.Length - 1) / 1000 + 1;
        for (int i = 0; i < pageCount; i++)
        {
            var personList = persons.Skip(i * 1000).Take(1000).ToArray();
            var values = personList.Select(p =>
                $"({p.Id},'{p.Name}',{p.Age},{(p.CreateTime.HasValue ? $"'{p.CreateTime:yyyy-MM-dd HH:mm:ss}'" : "NULL")},{(int) p.Sex})");
            var insertSql =
                $"INSERT INTO dbo.Person(Id,Name,Age,CreateTime,Sex)VALUES{string.Join(",", values)}";
            using (var com = new SqlCommand(insertSql, con))
            {
                com.ExecuteNonQuery();
            }
        }
    }
}

3.`SqlBulkCopy`插入方案

public static void BulkCopy(IEnumerable<Person> persons)
{
    using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;"))
    {
        con.Open();
        var table = new DataTable();
        table.Columns.AddRange(new []
        {
            new DataColumn("Id", typeof(long)), 
            new DataColumn("Name", typeof(string)), 
            new DataColumn("Age", typeof(int)), 
            new DataColumn("CreateTime", typeof(DateTime)), 
            new DataColumn("Sex", typeof(int)), 
        });
        foreach (var p in persons)
        {
            table.Rows.Add(new object[] {p.Id, p.Name, p.Age, p.CreateTime, (int) p.Sex});
        }

        using (var copy = new SqlBulkCopy(con))
        {
            copy.DestinationTableName = "Person";
            copy.WriteToServer(table);
        }
    }
}

3.三種方案速度對比

方案	數量	時間
Insert	1千條	145.4351ms
BatchInsert	1千條	103.9061ms
SqlBulkCopy	1千條	7.021ms

Insert	1萬條	1501.326ms
BatchInsert	1萬條	850.6274ms
SqlBulkCopy	1萬條	30.5129ms

Insert	10萬條	13875.4934ms
BatchInsert	10萬條	8278.9056ms
SqlBulkCopy	10萬條	314.8402ms

兩者插入效率對比，Insert明顯比SqlBulkCopy要慢太多，大概20~40倍性能差距，下面我們將SqlBulkCopy封裝一下，讓批量插入更加方便

二、SqlBulkCopy封裝代碼

1.方法介紹

批量插入擴展方法簽名

方法	方法引數	介紹
BulkCopy		同步的批量插入方法
	SqlConnection connection	sql server 連接物件
	IEnumerable<T> source	需要批量插入的資料源
	string tableName = null	插入表名稱【為NULL默認為物體名稱】
	int bulkCopyTimeout = 30	批量插入超時時間
	int batchSize = 0	寫入資料庫一批數量【如果為0代表全部一次性插入】最合適數量【這取決于您的環境，尤其是行數和網路延遲，就個人而言，我將從BatchSize屬性設定為1000行開始，然后看看其性能如何，如果可行，那么我將使行數加倍（例如增加到2000、4000等），直到性能下降或超時，否則，如果超時發生在1000，那么我將行數減少一半（例如500），直到它起作用為止，】
	SqlBulkCopyOptions options = SqlBulkCopyOptions.Default	批量復制引數
	SqlTransaction externalTransaction = null	執行的事務物件
BulkCopyAsync		異步的批量插入方法
	SqlConnection connection	sql server 連接物件
	IEnumerable<T> source	需要批量插入的資料源
	string tableName = null	插入表名稱【為NULL默認為物體名稱】
	int bulkCopyTimeout = 30	批量插入超時時間
	int batchSize = 0	寫入資料庫一批數量【如果為0代表全部一次性插入】最合適數量【這取決于您的環境，尤其是行數和網路延遲，就個人而言，我將從BatchSize屬性設定為1000行開始，然后看看其性能如何，如果可行，那么我將使行數加倍（例如增加到2000、4000等），直到性能下降或超時，否則，如果超時發生在1000，那么我將行數減少一半（例如500），直到它起作用為止，】
	SqlBulkCopyOptions options = SqlBulkCopyOptions.Default	批量復制引數
	SqlTransaction externalTransaction = null	執行的事務物件

這個方法主要解決了兩個問題：
- 免去了手動構建DataTable或者IDataReader介面實作類，手動構建的轉換比較難以維護，如果修改欄位就得把這些地方都進行修改，特別是還需要將列舉型別特殊處理，轉換成他的基礎型別（默認int）
- 不用親自創建SqlBulkCopy物件，和配置資料庫列的映射，和一些屬性的配置
此方案也是在我公司中使用，以滿足公司的批量插入資料的需求，例如第三方的對賬資料
此方法使用的是Expression動態生成資料轉換函式，其效率和手寫的原生代碼差不多，和原生手寫代碼相比，多余的轉換損失很小【最大的性能損失都是在值型別拆裝箱上】
此方案和其他網上的方案有些不同的是：不是將List先轉換成DataTable，然后寫入SqlBulkCopy的，而是使用一個實作IDataReader的讀取器包裝List，每往SqlBulkCopy插入一行資料才會轉換一行資料
IDataReader方案和DataTable方案相比優點
- 效率高：DataTable方案需要先完全轉換后，才能交由SqlBulkCopy寫入資料庫，而IDataReader方案可以邊轉換邊交給SqlBulkCopy寫入資料庫（例如：10萬資料插入速度可提升30%）
- 占用記憶體少：DataTable方案需要先完全轉換后，才能交由SqlBulkCopy寫入資料庫，需要占用大量記憶體，而IDataReader方案可以邊轉換邊交給SqlBulkCopy寫入資料庫，無須占用過多記憶體
- 強大：因為是邊寫入邊轉換，而且EnumerableReader傳入的是一個迭代器，可以實作持續插入資料的效果

2.實作原理

① 物體Model與表映射

資料庫表代碼

CREATE TABLE [dbo].[Person](
	[Id] [BIGINT] NOT NULL,
	[Name] [VARCHAR](64) NOT NULL,
	[Age] [INT] NOT NULL,
	[CreateTime] [DATETIME] NULL,
	[Sex] [INT] NOT NULL,
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]

物體類代碼

public class Person
{
    public long Id { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
    public DateTime? CreateTime { get; set; }
    public Gender Sex { get; set; }
}

public enum Gender
{
    Man = 0,
    Woman = 1
}

創建欄位映射【如果沒有此欄位映射會導致資料填錯位置，如果型別不對還會導致報錯】【因為：沒有此欄位映射默認是按照列序號對應插入的】
創建映射使用的SqlBulkCopy型別的ColumnMappings屬性來完成，資料列與資料庫中列的映射

//創建批量插入物件
using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
{
    foreach (var column in ModelToDataTable<TModel>.Columns)
    {
        //創建欄位映射
        copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
    }
}

② 物體轉換成資料行

將資料轉換成資料行采用的是：反射+Expression來完成
- 其中反射是用于獲取撰寫Expression所需程式類，屬性等資訊
- 其中Expression是用于生成高效轉換函式
其中ModelToDataTable<TModel>型別利用了靜態泛型類特性，實作泛型引數的快取效果
在ModelToDataTable<TModel>的靜態建構式中，生成轉換函式，獲取需要轉換的屬性資訊，并存入靜態只讀欄位中，完成快取

③ 使用IDataReader插入資料的多載

EnumerableReader是實作了IDataReader介面的讀取類，用于將模型物件，在迭代器中讀取出來，并轉換成資料行，可供SqlBulkCopy讀取
SqlBulkCopy只會呼叫三個方法：GetOrdinal、Read、GetValue
- 其中GetOrdinal只會在首行讀取每個列所代表序號【需要填寫：SqlBulkCopy型別的ColumnMappings屬性】
- 其中Read方法是迭代到下一行，并呼叫ModelToDataTable<TModel>.ToRowData.Invoke()來將模型物件轉換成資料行object[]
- 其中GetValue方法是獲取當前行指定下標位置的值

3.完整代碼

擴展方法類

    public static class SqlConnectionExtension
    {
        /// <summary>
        /// 批量復制
        /// </summary>
        /// <typeparam name="TModel">插入的模型物件</typeparam>
        /// <param name="source">需要批量插入的資料源</param>
        /// <param name="connection">資料庫連接物件</param>
        /// <param name="tableName">插入表名稱【為NULL默認為物體名稱】</param>
        /// <param name="bulkCopyTimeout">插入超時時間</param>
        /// <param name="batchSize">寫入資料庫一批數量【如果為0代表全部一次性插入】最合適數量【這取決于您的環境，尤其是行數和網路延遲，就個人而言，我將從BatchSize屬性設定為1000行開始，然后看看其性能如何，如果可行，那么我將使行數加倍（例如增加到2000、4000等），直到性能下降或超時，否則，如果超時發生在1000，那么我將行數減少一半（例如500），直到它起作用為止，】</param>
        /// <param name="options">批量復制引數</param>
        /// <param name="externalTransaction">執行的事務物件</param>
        /// <returns>插入數量</returns>
        public static int BulkCopy<TModel>(this SqlConnection connection,
            IEnumerable<TModel> source,
            string tableName = null,
            int bulkCopyTimeout = 30,
            int batchSize = 0,
            SqlBulkCopyOptions options = SqlBulkCopyOptions.Default,
            SqlTransaction externalTransaction = null)
        {
            //創建讀取器
            using (var reader = new EnumerableReader<TModel>(source))
            {
                //創建批量插入物件
                using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
                {
                    //插入的表
                    copy.DestinationTableName = tableName ?? typeof(TModel).Name;
                    //寫入資料庫一批數量
                    copy.BatchSize = batchSize;
                    //超時時間
                    copy.BulkCopyTimeout = bulkCopyTimeout;
                    //創建欄位映射【如果沒有此欄位映射會導致資料填錯位置，如果型別不對還會導致報錯】【因為：沒有此欄位映射默認是按照列序號對應插入的】
                    foreach (var column in ModelToDataTable<TModel>.Columns)
                    {
                        //創建欄位映射
                        copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
                    }
                    //將資料批量寫入資料庫
                    copy.WriteToServer(reader);
                    //回傳插入資料數量
                    return reader.Depth;
                }
            }
        }

        /// <summary>
        /// 批量復制-異步
        /// </summary>
        /// <typeparam name="TModel">插入的模型物件</typeparam>
        /// <param name="source">需要批量插入的資料源</param>
        /// <param name="connection">資料庫連接物件</param>
        /// <param name="tableName">插入表名稱【為NULL默認為物體名稱】</param>
        /// <param name="bulkCopyTimeout">插入超時時間</param>
        /// <param name="batchSize">寫入資料庫一批數量【如果為0代表全部一次性插入】最合適數量【這取決于您的環境，尤其是行數和網路延遲，就個人而言，我將從BatchSize屬性設定為1000行開始，然后看看其性能如何，如果可行，那么我將使行數加倍（例如增加到2000、4000等），直到性能下降或超時，否則，如果超時發生在1000，那么我將行數減少一半（例如500），直到它起作用為止，】</param>
        /// <param name="options">批量復制引數</param>
        /// <param name="externalTransaction">執行的事務物件</param>
        /// <returns>插入數量</returns>
        public static async Task<int> BulkCopyAsync<TModel>(this SqlConnection connection,
            IEnumerable<TModel> source,
            string tableName = null,
            int bulkCopyTimeout = 30,
            int batchSize = 0,
            SqlBulkCopyOptions options = SqlBulkCopyOptions.Default,
            SqlTransaction externalTransaction = null)
        {
            //創建讀取器
            using (var reader = new EnumerableReader<TModel>(source))
            {
                //創建批量插入物件
                using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
                {
                    //插入的表
                    copy.DestinationTableName = tableName ?? typeof(TModel).Name;
                    //寫入資料庫一批數量
                    copy.BatchSize = batchSize;
                    //超時時間
                    copy.BulkCopyTimeout = bulkCopyTimeout;
                    //創建欄位映射【如果沒有此欄位映射會導致資料填錯位置，如果型別不對還會導致報錯】【因為：沒有此欄位映射默認是按照列序號對應插入的】
                    foreach (var column in ModelToDataTable<TModel>.Columns)
                    {
                        //創建欄位映射
                        copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
                    }
                    //將資料批量寫入資料庫
                    await copy.WriteToServerAsync(reader);
                    //回傳插入資料數量
                    return reader.Depth;
                }
            }
        }
    }

封裝的迭代器資料讀取器

    /// <summary>
    /// 迭代器資料讀取器
    /// </summary>
    /// <typeparam name="TModel">模型型別</typeparam>
    public class EnumerableReader<TModel> : IDataReader
    {
        /// <summary>
        /// 實體化迭代器讀取物件
        /// </summary>
        /// <param name="source">模型源</param>
        public EnumerableReader(IEnumerable<TModel> source)
        {
            _source = source ?? throw new ArgumentNullException(nameof(source));
            _enumerable = source.GetEnumerator();
        }

        private readonly IEnumerable<TModel> _source;
        private readonly IEnumerator<TModel> _enumerable;
        private object[] _currentDataRow = Array.Empty<object>();
        private int _depth;
        private bool _release;

        public void Dispose()
        {
            _release = true;
            _enumerable.Dispose();
        }

        public int GetValues(object[] values)
        {
            if (values == null) throw new ArgumentNullException(nameof(values));
            var length = Math.Min(_currentDataRow.Length, values.Length);
            Array.Copy(_currentDataRow, values, length);
            return length;
        }

        public int GetOrdinal(string name)
        {
            for (int i = 0; i < ModelToDataTable<TModel>.Columns.Count; i++)
            {
                if (ModelToDataTable<TModel>.Columns[i].ColumnName == name) return i;
            }

            return -1;
        }

        public long GetBytes(int ordinal, long dataIndex, byte[] buffer, int bufferIndex, int length)
        {
            if (dataIndex < 0) throw new Exception($"起始下標不能小于0！");
            if (bufferIndex < 0) throw new Exception("目標緩沖區起始下標不能小于0！");
            if (length < 0) throw new Exception("讀取長度不能小于0！");
            var numArray = (byte[])GetValue(ordinal);
            if (buffer == null) return numArray.Length;
            if (buffer.Length <= bufferIndex) throw new Exception("目標緩沖區起始下標不能大于目標緩沖區范圍！");
            var freeLength = Math.Min(numArray.Length - bufferIndex, length);
            if (freeLength <= 0) return 0;
            Array.Copy(numArray, dataIndex, buffer, bufferIndex, length);
            return freeLength;
        }

        public long GetChars(int ordinal, long dataIndex, char[] buffer, int bufferIndex, int length)
        {
            if (dataIndex < 0) throw new Exception($"起始下標不能小于0！");
            if (bufferIndex < 0) throw new Exception("目標緩沖區起始下標不能小于0！");
            if (length < 0) throw new Exception("讀取長度不能小于0！");
            var numArray = (char[])GetValue(ordinal);
            if (buffer == null) return numArray.Length;
            if (buffer.Length <= bufferIndex) throw new Exception("目標緩沖區起始下標不能大于目標緩沖區范圍！");
            var freeLength = Math.Min(numArray.Length - bufferIndex, length);
            if (freeLength <= 0) return 0;
            Array.Copy(numArray, dataIndex, buffer, bufferIndex, length);
            return freeLength;
        }

        public bool IsDBNull(int i)
        {
            var value = https://www.cnblogs.com/liuzhenliang/p/GetValue(i);
            return value == null || value is DBNull;
        }
        public bool NextResult()
        {
            //移動到下一個元素
            if (!_enumerable.MoveNext()) return false;
            //行層+1
            Interlocked.Increment(ref _depth);
            //得到資料行
            _currentDataRow = ModelToDataTable.ToRowData.Invoke(_enumerable.Current);
            return true;
        }

        public byte GetByte(int i) => (byte)GetValue(i);
        public string GetName(int i) => ModelToDataTable.Columns[i].ColumnName;
        public string GetDataTypeName(int i) => ModelToDataTable.Columns[i].DataType.Name;
        public Type GetFieldType(int i) => ModelToDataTable.Columns[i].DataType;
        public object GetValue(int i) => _currentDataRow[i];
        public bool GetBoolean(int i) => (bool)GetValue(i);
        public char GetChar(int i) => (char)GetValue(i);
        public Guid GetGuid(int i) => (Guid)GetValue(i);
        public short GetInt16(int i) => (short)GetValue(i);
        public int GetInt32(int i) => (int)GetValue(i);
        public long GetInt64(int i) => (long)GetValue(i);
        public float GetFloat(int i) => (float)GetValue(i);
        public double GetDouble(int i) => (double)GetValue(i);
        public string GetString(int i) => (string)GetValue(i);
        public decimal GetDecimal(int i) => (decimal)GetValue(i);
        public DateTime GetDateTime(int i) => (DateTime)GetValue(i);
        public IDataReader GetData(int i) => throw new NotSupportedException();
        public int FieldCount => ModelToDataTable.Columns.Count;
        public object this[int i] => GetValue(i);
        public object this[string name] => GetValue(GetOrdinal(name));
        public void Close() => Dispose();
        public DataTable GetSchemaTable() => ModelToDataTable.ToDataTable(_source);
        public bool Read() => NextResult();
        public int Depth => _depth;
        public bool IsClosed => _release;
        public int RecordsAffected => 0;
    }

模型物件轉資料行工具類

    /// <summary>
    /// 物件轉換成DataTable轉換類
    /// </summary>
    /// <typeparam name="TModel">泛型型別</typeparam>
    public static class ModelToDataTable<TModel>
    {
        static ModelToDataTable()
        {
            //如果需要剔除某些列可以修改這段代碼
            var propertyList = typeof(TModel).GetProperties().Where(w => w.CanRead).ToArray();
            Columns = new ReadOnlyCollection<DataColumn>(propertyList
                .Select(pr => new DataColumn(pr.Name, GetDataType(pr.PropertyType))).ToArray());
            //生成物件轉資料行委托
            ToRowData = https://www.cnblogs.com/liuzhenliang/p/BuildToRowDataDelegation(typeof(TModel), propertyList);
        }

        /// 
        /// 構建轉換成資料行委托
        /// 
        ///


三、測驗封裝代碼
1.測驗代碼
創表代碼
CREATE TABLE [dbo].[Person](
	[Id] [BIGINT] NOT NULL,
	[Name] [VARCHAR](64) NOT NULL,
	[Age] [INT] NOT NULL,
	[CreateTime] [DATETIME] NULL,
	[Sex] [INT] NOT NULL,
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]

物體類代碼

定義的物體的屬性名稱需要和SqlServer列名稱型別對應

public class Person
{
    public long Id { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
    public DateTime? CreateTime { get; set; }
    public Gender Sex { get; set; }
}

public enum Gender
{
    Man = 0,
    Woman = 1
}

測驗方法
//生成10萬條資料
var persons = new Person[100000];
var random = new Random();
for (int i = 0; i < persons.Length; i++)
{
    persons[i] = new Person
    {
        Id = i + 1,
        Name = "張三" + i,
        Age = random.Next(1, 128),
        Sex = (Gender)random.Next(2),
        CreateTime = random.Next(2) == 0 ? null : (DateTime?) DateTime.Now.AddSeconds(i)
    };
}

//創建資料庫連接
using (var conn = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;"))
{
    conn.Open();
    var sw = Stopwatch.StartNew();
    //批量插入資料
    var qty = conn.BulkCopy(persons);
    sw.Stop();
    Console.WriteLine(sw.Elapsed.TotalMilliseconds + "ms");
}

執行批量插入結果
226.4767ms
請按任意鍵繼續. . .


四、代碼下載
GitHub代碼地址：https://github.com/liu-zhen-liang/PackagingComponentsSet/tree/main/SqlBulkCopyComponents






        
      轉載請註明出處，本文鏈接：https://www.uj5u.com/net/231441.html
      標籤：C# 
      上一篇：wpf對顯卡要求很高嗎?
下一篇：.NET 開源匯入匯出庫 Magicodes.IE 2.5發布

SQL Server 批量插入資料方案 SqlBulkCopy 的簡單封裝，讓批量插入更方便

一、Sql Server插入方案介紹

1.普通的Insert插入方法

2.拼接BatchInsert插入陳述句

3.SqlBulkCopy插入方案

3.三種方案速度對比

二、SqlBulkCopy封裝代碼

1.方法介紹

2.實作原理

① 物體Model與表映射

② 物體轉換成資料行

③ 使用IDataReader插入資料的多載

3.完整代碼

三、測驗封裝代碼

1.測驗代碼

四、代碼下載

1.普通的`Insert`插入方法

2.拼接`BatchInsert`插入陳述句

3.`SqlBulkCopy`插入方案