在SQL Server中使用SQLBulkCopy批量插入数据

在.Net1.1中无论是对于批量插入整个DataTable中的所有数据到数据库中，还是进行不同数据源之间的迁移，都不是很方便。而在.Net2.0中，SQLClient命名空间下增加了几个新类帮助我们通过DataTable或DataReader批量迁移数据。数据源可以来自关 系数据库或者XML文件，甚至WebService返回结果。其中最重要的一个类就是SqlBulkCopy类，使用它可以很方便的帮助我们把数据源的数 据迁移到目标数据库中。

SQL Server  SqlBulkCopy 的使用，高效率批量插入数据。方法笔记，未测。

class SqlBulk
    {
        /// <summary>
        /// 向数据库目标表中插入数据
        /// </summary>
        /// <param name="sourceTable">源数据DataTable</param>
        /// <param name="connectStr">数据库连接字符串</param>
        /// <param name="destinationTableName">数据库中的目标表名</param>
        /// <returns>是否成功</returns>
        public bool SqlBulkInsert(DataTable sourceTable, string connectStr, string destinationTableName)
        {
            try
            {
                SqlConnection conn = new SqlConnection(connectStr);
                SqlBulkInsert(sourceTable, conn, destinationTableName);
            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {

            }
            return true;
        }
        /// <summary>
        /// 向数据库目标表中插入数据
        /// </summary>
        /// <typeparam name="T">源数据的基本数据类型</typeparam>
        /// <param name="sourceList">源数据列表</param>
        /// <param name="connectStr">数据库连接字符串</param>
        /// <param name="destinationTableName">数据库中的目标表名</param>
        /// <returns>是否成功</returns>
        public bool SqlBulkInsert<T>(List<T> sourceList, string connectStr, string destinationTableName) where T:new ()
        {
            try
            {
                SqlConnection conn = new SqlConnection(connectStr);
                DataTable sourceTable=new DataTable();
                sourceTable = ListToDataTable<T>(sourceList, sourceTable);
                SqlBulkInsert(sourceTable, conn, destinationTableName);
            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {

            }
            return true;
        }
        /// <summary>
        /// 向数据库目标表中插入数据
        /// </summary>
        /// <typeparam name="T">源数据的基本数据类型</typeparam>
        /// <param name="sourceList">源数据列表</param>
        /// <param name="conn">未打开的数据库连接</param>
        /// <param name="destinationTableName">数据库中的目标表名</param>
        /// <returns>是否成功</returns>
        public bool SqlBulkInsert<T>(List<T> sourceList, SqlConnection conn, string destinationTableName) where T:new ()
        {
            try
            {
                DataTable sourceTable = new DataTable();
                sourceTable = ListToDataTable<T>(sourceList, sourceTable);
                SqlBulkInsert(sourceTable, conn, destinationTableName);

            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {

            }
            return true;
        }
        /// <summary>
        /// 向数据库目标表中插入数据
        /// </summary>
        /// <param name="sourceTable">源数据DataTable</param>
        /// <param name="conn">未打开的数据库连接</param>
        /// <param name="destinationTableName">数据库中的目标表名</param>
        /// <returns>是否成功</returns>
        public bool SqlBulkInsert(DataTable sourceTable, SqlConnection conn, string destinationTableName)
        {
            DataSet ds = new DataSet();
            using (SqlConnection _conn=new SqlConnection(conn.ConnectionString))
            {
                _conn.Open();
                SqlCommand cmd = new SqlCommand("SELECT TOP 0 * FROM " + destinationTableName, conn);
                SqlDataAdapter da = new SqlDataAdapter(cmd);
                da.Fill(ds);
                da.Dispose();
            }
            foreach (DataColumn col in ds.Tables[0].Columns)
            {
                if (!sourceTable.Columns.Contains(col.ColumnName))
                {
                    throw new Exception("源数据表结构与目标表结构不对应");
                }
            }
            SqlBulkCopy sbc = new SqlBulkCopy(conn);
            try
            {
                sbc.BulkCopyTimeout = 1800;
                sbc.DestinationTableName = destinationTableName;
                sbc.BatchSize = 1000;
                sbc.NotifyAfter = 1;
                sbc.SqlRowsCopied += new SqlRowsCopiedEventHandler(sbc_SqlRowsCopied);
                foreach (DataColumn col in ds.Tables[0].Columns)
                {
                    sbc.ColumnMappings.Add(col.ColumnName, col.ColumnName);
                }
                conn.Open();
                sbc.WriteToServer(sourceTable);
                sbc.Close();
            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {
                sbc.Close();
                conn.Close();
            }
            return true;
        }

        void sbc_SqlRowsCopied(object sender, SqlRowsCopiedEventArgs e)
        {
            //Console.WriteLine(e.RowsCopied + "rows are copierd");
        }
        /// <summary>
        /// 根据目标数据表结构，将List中的数据尽量填充到目标数据表
        /// </summary>
        /// <typeparam name="T">数据类型</typeparam>
        /// <param name="list">数据列表</param>
        /// <param name="dt">数据表</param>
        /// <returns>返回数据表</returns>
        private DataTable ListToDataTable<T>(List<T> list, DataTable dt) where T : new()
        {
            dt.Clear();
            foreach (var item in list)
            {
                DataRow dr = dt.NewRow();
                foreach (var prop in typeof(T).GetProperties())
                {
                    if (dt.Columns.Contains(prop.Name))
                    {
                        dr[prop.Name] = prop.GetValue(item, null);
                    }
                }
                dt.Rows.Add(dr);
            }
            return dt;
        }
    }

使用SqlBulkCopy类进行数据插入其原理是采用了SQL Server的BCP协议进行数据的批量复制。这里我们先要建好一个DataTable（最好是通过DataAdapter来灌数据得到，因为这样出来的DataTable就已经有跟数据表相同的列定义，可以免去之后Mapping Column的步骤），把要插入的数据加进这个DataTable中，然后用SqlBulkCopy的实例来插入到数据库中。经过测试，SqlBulkCopy方法比直接用Sql语句插入数据的效率高出将近25倍。
 
　　另外批量导入SQL、MYSQL等数据是同样的for循环，使用拼出来的sql或者使用参数的方式传递或者使用事务等不同方式的传递效率都不同。如果不使用SqlBulkCopy的方式的话，我测试下来做快递是用一次事务来操作为最快。因为10000次的循环如果是每次提交，那么都有链接和停止数据库的操作，或者说他包含了1000次的小事务处理。如果外面就一个事务的话效率肯定会高。