C# 之 批量插入數據到 SQLServer 中

　　建立一個用來測試的數據庫和表，爲了讓插入數據更快，表中主鍵採用的是GUID，表中沒有建立任何索引。GUID必然是比自增加要快。而若是存在索引的狀況下，每次插入記錄都會進行索引重建，這是很是耗性能的。若是表中無可避免的存在索引，咱們能夠經過先刪除索引，而後批量插入，最後再重建索引的方式來提升效率。

create database CarSYS;    
go    
use CarSYS;    
go 
CREATE TABLE Product(
Id UNIQUEIDENTIFIER PRIMARY KEY,
NAME VARCHAR(50) NOT NULL,
Price DECIMAL(18,2) NOT NULL
)

　　經過SQL腳原本插入數據，常見以下四種方式。

　　1、經過 INSERT INTO tname (...) VALUES (...) 逐條插入，性能最差，不建議使用。

　　示例：

#region 方式一
        static void InsertOne()
        {
            Console.WriteLine("採用一條一條插入的方式實現");
            Stopwatch sw = new Stopwatch();

            using (SqlConnection conn = new SqlConnection(StrConnMsg)) //using中會自動Open和Close 鏈接。
            {
                string sql = "INSERT INTO Product(Id,Name,Price) VALUES(newid(),@p,@d)";
                conn.Open();
                for (int i = 0; i < totalRow; i++)
                {
                    using (SqlCommand cmd = new SqlCommand(sql, conn))
                    {
                        cmd.Parameters.AddWithValue("@p", "商品" + i);
                        cmd.Parameters.AddWithValue("@d", i);
                        sw.Start();
                        cmd.ExecuteNonQuery();
                        Console.WriteLine(string.Format("插入一條記錄，已耗時{0}毫秒", sw.ElapsedMilliseconds));
                    }
                    if (i == getRow)
                    {
                        sw.Stop();
                        break;
                    }
                }
            }
            Console.WriteLine(string.Format("插入{0}條記錄，每{4}條的插入時間是{1}毫秒,預估總得插入時間是{2}毫秒，{3}分鐘",
 totalRow, sw.ElapsedMilliseconds, ((sw.ElapsedMilliseconds / getRow) * totalRow), GetMinute((sw.ElapsedMilliseconds / getRow * totalRow)), getRow));
        }
        static int GetMinute(long l)
        {
            return (Int32)l / 60000;
        } 
        #endregion

　　運行結果以下：

 　　

　　結果顯示插入100w條記錄，預計須要50分鐘時間，每插入一條記錄大概須要3毫秒左右。

　　2、使用 insert bulk

BULK INSERT [ [ 'database_name'.][ 'owner' ].]{ 'table_name' FROM 'data_file' }       
    WITH  (   
            [ BATCHSIZE [ = batch_size ] ],       
            [ CHECK_CONSTRAINTS ],           
            [ CODEPAGE [ = 'ACP' | 'OEM' | 'RAW' | 'code_page' ] ],   
            [ DATAFILETYPE [ = 'char' | 'native'| 'widechar' | 'widenative' ] ],               
            [ FIELDTERMINATOR [ = 'field_terminator' ] ],   
            [ FIRSTROW [ = first_row ] ],   
            [ FIRE_TRIGGERS ],   
            [ FORMATFILE = 'format_file_path' ],   
            [ KEEPIDENTITY ],   
            [ KEEPNULLS ],   
            [ KILOBYTES_PER_BATCH [ = kilobytes_per_batch ] ],      
            [ LASTROW [ = last_row ] ],   
            [ MAXERRORS [ = max_errors ] ],   
            [ ORDER ( { column [ ASC | DESC ] } [ ,...n ] ) ],     
            [ ROWS_PER_BATCH [ = rows_per_batch ] ],   
            [ ROWTERMINATOR [ = 'row_terminator' ] ],             
            [ TABLOCK ],   
    )  

　　相關參數說明：

BULK INSERT    
   [ database_name . [ schema_name ] . | schema_name . ] [ table_name | view_name ]    
      FROM 'data_file'    
     [ WITH    
        (    
   [ [ , ] BATCHSIZE = batch_size ]    --BATCHSIZE指令來設置在單個事務中能夠插入到表中的記錄的數量   
   [ [ , ] CHECK_CONSTRAINTS ]     --指定在大容量導入操做期間，必須檢查全部對目標表或視圖的約束。若沒有 CHECK_CONSTRAINTS 選項，則全部 CHECK 和 FOREIGN KEY 約束都將被忽略，而且在此操做以後表的約束將標記爲不可信。   
   [ [ , ] CODEPAGE = { 'ACP' | 'OEM' | 'RAW' | 'code_page' } ]  --指定該數據文件中數據的代碼頁   
   [ [ , ] DATAFILETYPE =    
      { 'char' | 'native'| 'widechar' | 'widenative' } ]  --指定 BULK INSERT 使用指定的數據文件類型值執行導入操做。   
   [ [ , ] FIELDTERMINATOR = 'field_terminator' ]  --標識分隔內容的符號   
   [ [ , ] FIRSTROW = first_row ]    --指定要加載的第一行的行號。默認值是指定數據文件中的第一行   
   [ [ , ] FIRE_TRIGGERS ]     --是否啓動觸發器   
   [ [ , ] FORMATFILE = 'format_file_path' ]    
   [ [ , ] KEEPIDENTITY ]   --指定導入數據文件中的標識值用於標識列   
   [ [ , ] KEEPNULLS ]    --指定在大容量導入操做期間空列應保留一個空值，而不插入用於列的任何默認值   
   [ [ , ] KILOBYTES_PER_BATCH = kilobytes_per_batch ]    
   [ [ , ] LASTROW = last_row ]   --指定要加載的最後一行的行號   
   [ [ , ] MAXERRORS = max_errors ]   --指定容許在數據中出現的最多語法錯誤數，超過該數量後將取消大容量導入操做。   
   [ [ , ] ORDER ( { column [ ASC | DESC ] } [ ,...n ] ) ]  --指定數據文件中的數據如何排序   
   [ [ , ] ROWS_PER_BATCH = rows_per_batch ]    
   [ [ , ] ROWTERMINATOR = 'row_terminator' ]   --標識分隔行的符號   
   [ [ , ] TABLOCK ]     --指定爲大容量導入操做持續時間獲取一個表級鎖   
   [ [ , ] ERRORFILE = 'file_name' ]   --指定用於收集格式有誤且不能轉換爲 OLE DB 行集的行的文件。   
        )]

　　

#region 方式二
        static void InsertTwo()
        {
            Console.WriteLine("使用Bulk插入的實現方式");
            Stopwatch sw = new Stopwatch();
            DataTable dt = GetTableSchema(); 

            using (SqlConnection conn = new SqlConnection(StrConnMsg))
            {
                SqlBulkCopy bulkCopy = new SqlBulkCopy(conn);
                bulkCopy.DestinationTableName = "Product";
                bulkCopy.BatchSize = dt.Rows.Count;
                conn.Open();
                sw.Start();

                for (int i = 0; i < totalRow;i++ )
                {
                    DataRow dr = dt.NewRow();
                    dr[0] = Guid.NewGuid();
                    dr[1] = string.Format("商品", i);
                    dr[2] = (decimal)i;
                    dt.Rows.Add(dr);
                }
                    if (dt != null && dt.Rows.Count != 0)
                    {
                        bulkCopy.WriteToServer(dt);
                        sw.Stop();
                    }
                    Console.WriteLine(string.Format("插入{0}條記錄共花費{1}毫秒，{2}分鐘", totalRow, sw.ElapsedMilliseconds, GetMinute(sw.ElapsedMilliseconds)));
            }
        }
        static DataTable GetTableSchema()
        {
            DataTable dt = new DataTable();
            dt.Columns.AddRange(new DataColumn[] {   
        new DataColumn("Id",typeof(Guid)),  
        new DataColumn("Name",typeof(string)),  
        new DataColumn("Price",typeof(decimal))});
            return dt;
        }
        #endregion

　　運行結果以下：

　　

　　插入100w條記錄才8s多，推薦。

　　打開Sqlserver Profiler跟蹤，會發現執行的是以下語句：

insert bulk Product ([Id] UniqueIdentifier, [NAME] VarChar(50) COLLATE Chinese_PRC_CI_AS, [Price] Decimal(18,2))

　　3、使用 INSERT INTO xx select...

INSERT INTO Product(Id,Name,Price)
 SELECT NEWID(),'牛欄1段',160 
 UNION ALL 
 SELECT NEWID(),'牛欄2段',180
 UNION ALL
......

　　從sqlserver 2008起開始支持TVPs。建立緩存表ProductTemp ，執行以下SQL：

CREATE TYPE ProductTemp AS  TABLE(
Id UNIQUEIDENTIFIER PRIMARY KEY,
NAME VARCHAR(50) NOT NULL,
Price DECIMAL(18,2) NOT NULL
)

　　執行完成以後，會發如今數據庫CarSYS下面多了一張緩存表ProductTemp

　　

　　

　　可見插入100w條記錄共花費了11秒多，此方式麻煩，一次性可以使用。　　

　　4、拼接SQL

INSERT INTO Product(Id,Name,Price) VALUES
(newid(),'牛欄1段',160)
,(newid(),'牛欄2段',260)
......

 　　示例：此種方法在C#中有限制，一次性只能批量插入1000條，因此就得分段進行插入。

#region 方式四
        static void InsertFour()
        {
            Console.WriteLine("採用拼接批量SQL插入的方式實現");
            Stopwatch sw = new Stopwatch();

            using (SqlConnection conn = new SqlConnection(StrConnMsg)) //using中會自動Open和Close 鏈接。
            {
                conn.Open();
                sw.Start();
                for (int j = 0; j < totalRow / getRow;j++ )
                {
                    StringBuilder sb = new StringBuilder();
                    sb.Append("INSERT INTO Product(Id,Name,Price) VALUES");
                    using (SqlCommand cmd = new SqlCommand())
                    {
                       
                        for (int i = 0; i < getRow; i++)
                        {
                            sb.AppendFormat("(newid(),'商品{0}',{0}),", j*i+i);
                        }
                        cmd.Connection = conn;
                        cmd.CommandText = sb.ToString().TrimEnd(',');
                        cmd.ExecuteNonQuery();
                    }
                }
                sw.Stop();
                Console.WriteLine(string.Format("插入{0}條記錄，共耗時{1}毫秒",totalRow,sw.ElapsedMilliseconds));
            }
        }
        #endregion

　　運行結果以下：

　　

　　能夠看到大概花費了10分鐘。雖然在方式一的基礎上，性能有了較大的提高，可是顯然仍是不夠快。

　　總結：大數據批量插入方式一和方式四儘可能避免使用，而方式二和方式三都是很是高效的批量插入數據方式。其都是經過構建DataTable的方式插入的，而咱們知道DataTable是存在內存中的，因此當數據量特別特別大，大到內存中沒法一次性存儲的時候，能夠分段插入。好比須要插入9千萬條數據，能夠分紅9段進行插入，一次插入1千萬條。而在for循環中直接進行數據庫操做，咱們是應該儘可能避免的。每一次數據庫的鏈接、打開和關閉都是比較耗時的，雖然在C#中存在數據庫鏈接池，也就是當咱們使用using或者conn.Close()，進行釋放鏈接時，其實並無真正關閉數據庫鏈接，它只是讓鏈接以相似於休眠的方式存在，當再次操做的時候，會從鏈接池中找一個休眠狀態的鏈接，喚醒它，這樣能夠有效的提升併發能力，減小鏈接損耗。而鏈接池中的鏈接數，咱們都是能夠配置的。