C# & SQL Server大数据量插入方式对比

以下内容大部分来自：

http://blog.csdn.net/tjvictor/article/details/4360030

部分内容出自互联网，实验结果为亲测。

      最近自己开发一个向数据库中插入大量历史数据的函数库，需要解决一个大数据量插入的效率问题。不用分析，我知道如果采取逐条数据插入的方式，那么效率肯定很低，光是那么多循环就知道很慢了。于是乎，我找到了上篇博客，知道了BulkCopy和TVPs方式。为了更好的了解其效率，我自己动手亲测了一下效果，测试的数据库位于本机。

（1）方式1：循环插入

        public static void NormalInerst(String connString)
        {
            Console.WriteLine("使用NNormalInerst方式：");
            Stopwatch sw = new Stopwatch();
            SqlConnection sqlConn = new SqlConnection(connString);
            SqlCommand sqlCmd = new SqlCommand();
            sqlCmd.CommandText = String.Format("insert into BulkTestTable(Id,UserName,Pwd)values(@p0,@p1,@p2)");
            sqlCmd.Parameters.Add("@p0", SqlDbType.Int);
            sqlCmd.Parameters.Add("@p1", SqlDbType.NVarChar);
            sqlCmd.Parameters.Add("@p2", SqlDbType.VarChar);
            sqlCmd.CommandType = CommandType.Text;
            sqlCmd.Connection = sqlConn;
            sqlConn.Open();
            try
            {
                for (int i = 0, j = 0; i < 10; ++i )
                {
                    for (j = i * 10000; j < (i + 1) * 10000; ++j )
                    {
                        sqlCmd.Parameters["@p0"].Value = j;
                        sqlCmd.Parameters["@p1"].Value = String.Format("User-{0}", i * j);
                        sqlCmd.Parameters["@p2"].Value = String.Format("Pwd-{0}", i * j);
                        sw.Start();
                        sqlCmd.ExecuteNonQuery();
                        sw.Stop();
                    }
                    
                    Console.WriteLine("第{0}次插入{1}条数据耗时：{2}", (i + 1), dataScale, sw.ElapsedMilliseconds);
                    sw.Reset();
                }
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                sqlConn.Close();
            }
        }

该方式的效率极低，运行时间很长，我这里就不给出结果了，有兴趣可以自己粘贴试一下。PS:其中的数据规模应该是dataScale而不是10000，不过总是还是慢。

（2）方式2：使用BulkCopy

        public static void BulkInerst(String connString)
        {
            Console.WriteLine("使用BulkInerst方式：");
            Stopwatch sw = new Stopwatch();
            
            String strDel = "delete from BulkTestTable";
            float millTime = 0;
            for (int multiply = 0; multiply < 10; multiply++)
            {
                DataTable dt = GetTableSchema();
                for (int count = multiply * dataScale; count < (multiply + 1) * dataScale; count++)
                {
                    DataRow r = dt.NewRow();
                    r[0] = count;
                    r[1] = string.Format("User-{0}", count * multiply);
                    r[2] = string.Format("Pwd-{0}", count * multiply);
                    dt.Rows.Add(r);
                }

                SqlConnection sqlConn = new SqlConnection(connString);
                SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn);
                bulkCopy.DestinationTableName = "BulkTestTable";
                bulkCopy.BatchSize = dt.Rows.Count;

                sw.Reset();
                sw.Start();
                try
                {
                    sqlConn.Open();
                    if (dt != null && dt.Rows.Count != 0)
                        bulkCopy.WriteToServer(dt);
                }
                catch (Exception ex)
                {
                    throw ex;
                }
                finally
                {
                    sqlConn.Close();
                    if (bulkCopy != null)
                        bulkCopy.Close();
                } 
                sw.Stop();

                Console.WriteLine("第{0}次插入{1}条数据耗时：{2}", (multiply + 1), dataScale, sw.ElapsedMilliseconds);
                millTime += sw.ElapsedMilliseconds;
            }
            Console.WriteLine("总耗时：{0}毫秒，平均耗时：{1}毫秒", millTime, millTime / 10);
            SqlConnection sqlConn2 = new SqlConnection(connString);
            SqlCommand sqlCmd = new SqlCommand(strDel, sqlConn2);
            try
            {
                sqlConn2.Open();
                sqlCmd.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                sqlConn2.Close();
            }
            Console.WriteLine("Done!"); 
        }

（3）方式3：使用TVPs

        public static void TVPsInerst(String connString)
        {
            Console.WriteLine("使用TVPsInerst方式：");
            Stopwatch sw = new Stopwatch();
            SqlConnection sqlConn = new SqlConnection(connString);
            String strSQL = "insert into BulkTestTable (Id,UserName,Pwd)" +
                " SELECT nc.Id, nc.UserName,nc.Pwd" +
                " FROM @NewBulkTestTvp AS nc";
            String strDel = "delete from BulkTestTable";
            float millTime = 0;

            for (int multiply = 0; multiply < 10; multiply++)
            {
                DataTable dt = GetTableSchema();
                for (int count = multiply * dataScale; count < (multiply + 1) * dataScale; count++)
                {
                    DataRow r = dt.NewRow();
                    r[0] = count;
                    r[1] = string.Format("User-{0}", count * multiply);
                    r[2] = string.Format("Pwd-{0}", count * multiply);
                    dt.Rows.Add(r);
                }

                sw.Reset();
                sw.Start();
                SqlCommand cmd = new SqlCommand(strSQL, sqlConn);
                SqlParameter catParam = cmd.Parameters.AddWithValue("@NewBulkTestTvp", dt);
                catParam.SqlDbType = SqlDbType.Structured;
                catParam.TypeName = "dbo.BulkUDT";
                try
                {
                    sqlConn.Open();
                    if (dt != null && dt.Rows.Count != 0)
                    {
                        cmd.ExecuteNonQuery();
                    }
                }
                catch (Exception ex)
                {
                    throw ex;
                }
                finally
                {
                    sqlConn.Close();
                }
                sw.Stop();

                Console.WriteLine("第{0}次插入{1}条数据耗时：{2}", (multiply + 1), dataScale, sw.ElapsedMilliseconds);
                millTime += sw.ElapsedMilliseconds;
            }
            Console.WriteLine("总耗时：{0}毫秒，平均耗时：{1}毫秒", millTime, millTime / 10);
            SqlCommand sqlCmd = new SqlCommand(strDel, sqlConn);
            try
            {
                sqlConn.Open();
                sqlCmd.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                sqlConn.Close();
            }
            Console.WriteLine("Done!");
        }

这里TVPs方式需要利用Visual Studio 2008采用的自定义数据表类型，这是一个比较新的东西。这里补充几个类型和函数，主要是为了检测数据库中是否存在数据表和数据表类型，如果不存在则进行创建。补充代码如下：

        public enum CheckType
        {
            isTable = 0,
            isType
        }

        protected static int dataScale = 100000;

        public static bool CheckExistsObject(String connString, String objectName, CheckType type)
        {
            String strSQL = "select COUNT(1) from sys.sysobjects where name='" + objectName + "'";
            switch (type)
            {
                case CheckType.isTable:
                    strSQL = "select COUNT(1) from sys.sysobjects where name='" + objectName + "'";
                    break;
                case CheckType.isType:
                    strSQL = "select COUNT(1) from sys.types where name='" + objectName + "'";
                    break;
                default:
                    break;
            }
            using (SqlConnection conn = new SqlConnection(connString)) 
            {
                conn.Open();
                SqlCommand cmd = new SqlCommand(strSQL, conn);
                int result = Convert.ToInt32(cmd.ExecuteScalar());
                if (0 == result)
                {
                    return false;
                }
            }

            return true;
        }

        public static bool CreateObject(String connString, String objectName, CheckType type)
        {
            String strSQL = "";
            switch (type)
            {
                case CheckType.isTable:
                    strSQL = "Create table " + objectName + " (Id int primary key, UserName nvarchar(32), Pwd varchar(16))";
                    break;
                case CheckType.isType:
                    strSQL = "CREATE TYPE " + objectName + " AS TABLE (Id int, UserName nvarchar(32), Pwd varchar(16))";
                    break;
                default:
                    break;
            }
            using (SqlConnection conn = new SqlConnection(connString))
            {
                conn.Open();
                SqlCommand cmd = new SqlCommand(strSQL, conn);
                cmd.ExecuteNonQuery();
            }

            return true;
        }
        public static DataTable GetTableSchema()
        {
            DataTable dt = new DataTable();
            dt.Columns.AddRange(new DataColumn[]{  
                    new DataColumn("Id",typeof(int)),  
                    new DataColumn("UserName",typeof(string)),  
                    new DataColumn("Pwd",typeof(string))});

            return dt;
        }

调用的方式就很好说了，参见如下测试代码：

        public static void Main(string[] args)
        {
            String conString = "Persist Security Info=False;User ID=sa;Password=scbj123!@#;Initial Catalog=testGR;Server=KLH-PC";
            String strType = "BulkUDT";
            String strTable = "BulkTestTable";
            if (!CheckExistsObject(conString, strType, CheckType.isType))
            {
                Console.WriteLine("类型{0}不存在", strType);
                if (CreateObject(conString, strType, CheckType.isType))
                {
                    Console.WriteLine("类型{0}创建成功！", strType);
                }
            }

            if (!CheckExistsObject(conString, strTable, CheckType.isTable))
            {
                Console.WriteLine("表格{0}不存在", strTable);
                if (CreateObject(conString, strTable, CheckType.isTable))
                {
                    Console.WriteLine("表格{0}创建成功！", strTable);
                }
            }
            Console.WriteLine("==================================================");

            //NormalInerst(conString);
            BulkInerst(conString);
            TVPsInerst(conString);
            
            Console.ReadKey();
        }

-------------------------------------------------------------------------------------------------

直接看效果对比：

<1>第一次运行

<2>第二次和第三次运行

这里考虑到了SQL Server自身缓存的原因，所以进行了多次测试，不过数据量没有变。可以从上述结果中看出：TVPs方式不愧是新出的啊，一代更比一代强！