LIOT/Framework/Bowin.Common/Linq/BulkCopyExtensions.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data.SqlClient;
using System.Data;
using System.Collections;
using System.Configuration;
using Bowin.Common.Data;
using Bowin.Common.Linq.DB;

namespace Bowin.Common.Linq
{
    public static class BulkCopyExtensions
    {

        public static bool ExecuteBulkCopy<T>(this IEnumerable<T> dataSource, string dbName, string tableName)
            where T : class
        {
            SqlConnection conn;

            if (SqlConnectionManager.IsGlobalConnectionStarted)
            {
                conn = SqlConnectionManager.GetConnection(dbName);
            }
            else
            {
                conn = new SqlConnection(ConfigurationManager.ConnectionStrings[dbName].ConnectionString);

                if (conn.State == ConnectionState.Closed)
                {
                    conn.Open();
                }
            }
            try
            {
                var dt = dataSource.ToTable(true);
                dt.TableName = tableName;
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = dt.TableName;
                    bcp.WriteToServer(dt);
                }
            }
            catch (Exception ex)
            {
                throw (ex);
                //return false;
            }
            finally
            {
                if (!SqlConnectionManager.IsGlobalConnectionStarted)
                {
                    conn.Close();
                }
            }
            return true;
        }

        public static bool ExecuteBulkCopy<T>(this IEnumerable<T> dataSource, SqlConnection db, string tableName)
            where T : class
        {
            try
            {
                var dt = dataSource.ToTable(true);
                dt.TableName = tableName;
                if (db.State == ConnectionState.Closed)
                {
                    db.Open();
                }
                using (SqlBulkCopy bcp = new SqlBulkCopy(db))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = dt.TableName;
                    bcp.WriteToServer(dt);
                }
                return true;
            }
            catch (Exception ex)
            {
                throw (ex);
                //return false;
            }
            finally
            {
                if (!SqlConnectionManager.IsGlobalConnectionStarted)
                {
                    db.Close();
                }
            }
        }


        /// <summary>
        /// 带事务的bcp方式批量插入数据库操作(注意要指定TableName属性)。
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="dt"></param>
        /// <returns></returns>
        public static bool ExecuteBulkCopy(SqlConnection conn, DataTable dt, bool needCommit = true)
        {
            bool result = true;
            if (conn.State == ConnectionState.Closed)
            {
                conn.Open();
            }

            var stran = conn.BeginTransaction();
            try
            {
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, stran))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = dt.TableName;
                    bcp.WriteToServer(dt);
                    if (needCommit)
                    {
                        stran.Commit();
                    }
                }
            }
            catch (Exception ex)
            {
                result = false;
                stran.Rollback();
                throw ex;
            }
            //finally
            //{
            //    conn.Close();
            //}


            return result;

        }

        /// <summary>
        /// 带事务的bcp方式批量插入数据库操作(注意要指定TableName属性)。
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="dt"></param>
        /// <returns></returns>
        public static bool ExecuteBulkCopy(SqlConnection conn, SqlTransaction stran, DataTable dt)
        {
            bool result = true;
            try
            {
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, stran))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = dt.TableName;
                    bcp.WriteToServer(dt);
                }
            }
            catch (Exception ex)
            {
                result = false;
                string err = ex.Message;
                throw ex;
            }
            return result;
        }

        public static void BluckTable(SqlConnection conn, params DataTable[] blukParams)
        {


            if (conn.State == ConnectionState.Closed)
            {
                conn.Open();
            }
            var stran = conn.BeginTransaction();
            try
            {
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, stran))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    foreach (var bluck in blukParams)
                    {
                        Bowin.Common.Linq.BulkCopyExtensions.ExecuteBulkCopy(conn, stran, bluck);
                    }
                    stran.Commit();
                }
            }
            catch (Exception ex)
            {
                stran.Rollback();
                throw ex;
            }
            //finally
            //{
            //    conn.Close();
            //}
        }

        /// <summary>
        /// 将序列的数据插入到临时表中
        /// </summary>
        /// <param name="tempTableName">临时表名称，包括前面的#号</param>
        /// <param name="dataSource">任意可枚举数据源</param>
        /// <param name="singleColumnName">若为简单类型的列表，需指定临时表中的字段名</param>
        /// <remarks>必须将该方法在显式事务中调用，否则无法保证临时表的有效性</remarks>
        public static void BulkCopyToTempTable<T>(this SqlConnection conn, string tempTableName, IEnumerable<T> dataSource, string singleColumnName = "")
        {
            DataTable dt;

            if (singleColumnName == "")
            {
                dt = dataSource.ToTable(true);
            }
            else
            {
                dt = new DataTable("dataResult");
                var columnType = typeof(T);
                if (columnType.FullName.StartsWith("System.Nullable"))
                {
                    columnType = ((System.Type)(typeof(T))).GetGenericArguments()[0];
                }
                dt.Columns.Add(singleColumnName, columnType);
                foreach (var data in dataSource)
                {
                    var row = dt.NewRow();
                    if (data != null)
                    {
                        row[singleColumnName] = data;
                    }
                    else
                    {
                        row[singleColumnName] = DBNull.Value;
                    }
                    dt.Rows.Add(row);
                }
            }

            string sql = "create table " + tempTableName + " (";

            if (singleColumnName != "")
            {
                //System.Data.Entity.Infrastructure.DbCompiledModel
                sql += "[" + singleColumnName + "] " + ConvertToDataTable.SqlDbTypeConvertor[typeof(T)];
            }
            else
            {
                foreach (DataColumn col in dt.Columns)
                {
                    sql += "[" + col.ColumnName + "] " + ConvertToDataTable.SqlDbTypeConvertor[col.DataType] + ",";
                }

                sql = sql.TrimEnd(',');
            }

            sql += ")";

            if (conn.State == ConnectionState.Closed)
            {
                conn.Open();
            }

            SqlCommand comm = conn.CreateCommand();
            comm.CommandText = sql;
            comm.CommandTimeout = 6000;
            comm.ExecuteNonQuery();

            try
            {
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = tempTableName;
                    bcp.WriteToServer(dt);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        /// <summary>
        /// 将序列的数据插入到临时表中
        /// </summary>
        /// <param name="tempTableName">临时表名称，包括前面的#号</param>
        /// <param name="dataSource">任意可枚举数据源</param>
        /// <param name="singleColumnName">若为简单类型的列表，需指定临时表中的字段名</param>
        /// <remarks>必须将该方法在显式事务中调用，否则无法保证临时表的有效性</remarks>
        public static void BulkCopyToTempTableWithTransation<T>(this SqlConnection conn, SqlTransaction tran, string tempTableName, IEnumerable<T> dataSource, string singleColumnName = "")
        {
            DataTable dt;

            if (singleColumnName == "")
            {
                dt = dataSource.ToTable(true);
            }
            else
            {
                dt = new DataTable("dataResult");
                var columnType = typeof(T);
                if (columnType.FullName.StartsWith("System.Nullable"))
                {
                    columnType = ((System.Type)(typeof(T))).GetGenericArguments()[0];
                }
                dt.Columns.Add(singleColumnName, columnType);
                foreach (var data in dataSource)
                {
                    var row = dt.NewRow();
                    if (data != null)
                    {
                        row[singleColumnName] = data;
                    }
                    else
                    {
                        row[singleColumnName] = DBNull.Value;
                    }
                    dt.Rows.Add(row);
                }
            }

            string sql = "create table " + tempTableName + " (";

            if (singleColumnName != "")
            {
                sql += "[" + singleColumnName + "] " + ConvertToDataTable.SqlDbTypeConvertor[typeof(T)];
            }
            else
            {
                foreach (DataColumn col in dt.Columns)
                {
                    sql += "[" + col.ColumnName + "] " + ConvertToDataTable.SqlDbTypeConvertor[col.DataType] + ",";
                }

                sql = sql.TrimEnd(',');
            }

            sql += ")";



            SqlCommand comm = new SqlCommand(sql, conn, tran);
            comm.CommandText = sql;
            comm.CommandTimeout = 6000;
            comm.ExecuteNonQuery();

            try
            {
                using (SqlBulkCopy bcp = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, tran))
                {
                    bcp.BulkCopyTimeout = 3600; //默认值 30秒
                    bcp.DestinationTableName = tempTableName;
                    bcp.WriteToServer(dt);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }
}