发 帖  
原厂入驻New
[问答]

几种数据库的大数据批量插入解决方法

59 大数据
分享
在之前只知道SqlServer支持数据批量插入,殊不知道Oracle、SQLite和MySql也是支持的,不过Oracle需要使用Orace.DataAccess驱动,今天就贴出几种数据库的批量插入解决方法。
   
0
2020-11-4 07:59:45   评论 分享淘帖 邀请回答
2个回答
首先说一下,IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider,此接口在前一篇文章中已经提到过了。
<img    /><img    /><img  />
  /// <summary>    /// 提供数据批量处理的方法。    /// </summary>    public interface IBatcherProvider : IProviderService    {        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        void Insert(DataTable dataTable, int batchSize = 10000);    } <img  />

View Code
一、SqlServer数据批量插入
    SqlServer的批量插入很简单,使用SqlBulkCopy就可以,以下是该类的实现:

<img    /><img    /><img  />
/// <summary>    /// 为 System.Data.SqlClient 提供的用于批量操作的方法。    /// </summary>    public sealed class MsSqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = (SqlConnection)ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    //给表名加上前后导符                    var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider>(), dataTable.TableName);                    using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)                        {                            DestinationTableName = tableName,                             BatchSize = batchSize                        })                    {                        //循环所有列,为bulk添加映射                        dataTable.EachColumn(c => bulk.ColumnMappings.Add(c.ColumnName, c.ColumnName), c => !c.AutoIncrement);                        bulk.WriteToServer(dataTable);                        bulk.Close();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }    }      SqlBulkCopy的ColumnMappings中列的名称受大小写敏感限制,因此在构造DataTable的时候应请注意列名要与表一致。     以上没有使用事务,使用事务在性能上会有一定的影响,如果要使用事务,可以设置SqlBulkCopyOptions.UseInternalTransaction。<img  />

View Code
二、Oracle数据批量插入
     System.Data.OracleClient不支持批量插入,因此只能使用Oracle.DataAccess组件来作为提供者。
<img    /><img    /><img  />
/// <summary>    /// Oracle.Data.Access 组件提供的用于批量操作的方法。    /// </summary>    public sealed class OracleAccessBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            //将一个DataTable的数据转换为数组的数组            var data = table.ToArray();            //设置ArrayBindCount属性            command.GetType().GetProperty(&quot;ArrayBindCount&quot;).SetValue(command, table.Rows.Count, null);            var syntax = database.Provider.GetService<ISyntaxProvider>();            for (var i = 0; i < table.Columns.Count; i++)            {                var column = table.Columns;                var parameter = database.Provider.DbProviderFactory.CreateParameter();                if (parameter == null)                {                    continue;                }                parameter.ParameterName = column.ColumnName;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = column.DataType.GetDbType();                parameter.Value = data;                if (names.Length > 0)                {                    names.Append(&quot;,&quot;);                    values.Append(&quot;,&quot;);                }                names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, column.ColumnName));                values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                command.Parameters.Add(parameter);            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
    以上最重要的一步,就是将DataTable转为数组的数组表示,即object[][],前数组的上标是列的个数,后数组是行的个数,因此循环Columns将后数组作为Parameter的值,也就是说,参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。
三、SQLite数据批量插入
     SQLite的批量插入只需开启事务就可以了,这个具体的原理不得而知。
<img    /><img    /><img  />
public sealed class SQLiteBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                DbTransaction transcation = null;                try                {                    connection.TryOpen();                    transcation = connection.BeginTransaction();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        var flag = new AssertFlag();                        dataTable.EachRow(row =>                            {                                var first = flag.AssertTrue();                                ProcessCommandParameters(dataTable, command, row, first);                                command.ExecuteNonQuery();                            });                    }                    transcation.Commit();                }                catch (Exception exp)                {                    if (transcation != null)                    {                        transcation.Rollback();                    }                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        private void ProcessCommandParameters(DataTable dataTable, DbCommand command, DataRow row, bool first)        {            for (var c = 0; c < dataTable.Columns.Count; c++)            {                DbParameter parameter;                //首次创建参数,是为了使用缓存                if (first)                {                    parameter = ServiceContext.Database.Provider.DbProviderFactory.CreateParameter();                    parameter.ParameterName = dataTable.Columns[c].ColumnName;                    command.Parameters.Add(parameter);                }                else                {                    parameter = command.Parameters[c];                }                parameter.Value = row[c];            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DataTable table)        {            var syntax = database.Provider.GetService<ISyntaxProvider>();            var names = new StringBuilder();            var values = new StringBuilder();            var flag = new AssertFlag();            table.EachColumn(column =>                {                    if (!flag.AssertTrue())                    {                        names.Append(&quot;,&quot;);                        values.Append(&quot;,&quot;);                    }                    names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName));                    values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                });            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
   四、MySql数据批量插入
<img    /><img    /><img  />
/// <summary>    /// 为 MySql.Data 组件提供的用于批量操作的方法。    /// </summary>    public sealed class MySqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            var types = new List<DbType>();            var count = table.Columns.Count;            var syntax = database.Provider.GetService<ISyntaxProvider>();            table.EachColumn(c =>                {                    if (names.Length > 0)                    {                        names.Append(&quot;,&quot;);                    }                    names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, c.ColumnName));                    types.Add(c.DataType.GetDbType());                });            var i = 0;            foreach (DataRow row in table.Rows)            {                if (i > 0)                {                    values.Append(&quot;,&quot;);                }                values.Append(&quot;(&quot;);                for (var j = 0; j < count; j++)                {                    if (j > 0)                    {                        values.Append(&quot;, &quot;);                    }                    var isStrType = IsStringType(types[j]);                    var parameter = CreateParameter(database.Provider, isStrType, types[j], row[j], syntax.ParameterPrefix, i, j);                    if (parameter != null)                    {                        values.Append(parameter.ParameterName);                        command.Parameters.Add(parameter);                    }                    else if (isStrType)                    {                        values.AppendFormat(&quot;&#39;{0}&#39;&quot;, row[j]);                    }                    else                    {                        values.Append(row[j]);                    }                }                values.Append(&quot;)&quot;);                i++;            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES {2}&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }        /// <summary>        /// 判断是否为字符串类别。        /// </summary>        /// <param name=&quot;dbType&quot;></param>        /// <returns></returns>        private bool IsStringType(DbType dbType)        {            return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength;        }        /// <summary>        /// 创建参数。        /// </summary>        /// <param name=&quot;provider&quot;></param>        /// <param name=&quot;isStrType&quot;></param>        /// <param name=&quot;dbType&quot;></param>        /// <param name=&quot;value&quot;></param>        /// <param name=&quot;parPrefix&quot;></param>        /// <param name=&quot;row&quot;></param>        /// <param name=&quot;col&quot;></param>        /// <returns></returns>        private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col)        {            //如果生成全部的参数,则速度会很慢,因此,只有数据类型为字符串(包含&#39;号)和日期型时才添加参数            if ((isStrType && value.ToString().IndexOf(&#39;/&#39;&#39;) != -1) || dbType == DbType.DateTime)            {                var name = string.Format(&quot;{0}p_{1}_{2}&quot;, parPrefix, row, col);                var parameter = provider.DbProviderFactory.CreateParameter();                parameter.ParameterName = name;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = dbType;                parameter.Value = value;                return parameter;            }            return null;        }    }<img  />

View Code
    MySql的批量插入,是将值全部写在语句的values里,例如,insert batcher(id, name) values(1, &#39;1&#39;, 2, &#39;2&#39;, 3, &#39;3&#39;, ........ 10, &#39;10&#39;)。

     五、测试
     接下来写一个测试用例来看一下使用批量插入的效果。
<img    /><img    /><img  />
[Test]        public void TestBatchInsert()        {            Console.WriteLine(TimeWatcher.Watch(() =>                InvokeTest(database =>                    {                        var table = new DataTable(&quot;Batcher&quot;);                        table.Columns.Add(&quot;Id&quot;, typeof(int));                        table.Columns.Add(&quot;Name1&quot;, typeof(string));                        table.Columns.Add(&quot;Name2&quot;, typeof(string));                        table.Columns.Add(&quot;Name3&quot;, typeof(string));                        table.Columns.Add(&quot;Name4&quot;, typeof(string));                        //构造100000条数据                        for (var i = 0; i < 100000; i++)                        {                            table.Rows.Add(i, i.ToString(), i.ToString(), i.ToString(), i.ToString());                        }                        //获取 IBatcherProvider                        var batcher = database.Provider.GetService<IBatcherProvider>();                        if (batcher == null)                        {                            Console.WriteLine(&quot;不支持批量插入。&quot;);                        }                        else                        {                            batcher.Insert(table);                        }                        //输出batcher表的数据量                        var sql = new SqlCommand(&quot;SELECT COUNT(1) FROM Batcher&quot;);                        Console.WriteLine(&quot;当前共有 {0} 条数据&quot;, database.ExecuteScalar(sql));                    })));        } <img  />

View Code
  以下表中列出了四种数据库生成10万条数据各耗用的时间

数据库

耗用时间

MsSql00:00:02.9376300
Oracle00:00:01.5155959
SQLite00:00:01.6275634
MySql00:00:05.4166891
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
<img    /><img    /><img  />
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// <summary>        /// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// </summary>        /// <param name=&quot;strTableName&quot;>数据库中对应的表名</param>        /// <param name=&quot;dtData&quot;>数据集</param>        public void SqlBulkCopyInsert(string strTableName, DataTable dtData)        {            string ConStr = connectionString;// 数据库连接字符串              try            {                using (SqlBulkCopy sqlRevdBulkCopy = new SqlBulkCopy(ConStr))//引用SqlBulkCopy                {                    sqlRevdBulkCopy.DestinationTableName = strTableName;//数据库中对应的表名                      sqlRevdBulkCopy.NotifyAfter = dtData.Rows.Count;//有几行数据                      sqlRevdBulkCopy.WriteToServer(dtData);//数据导入数据库                      sqlRevdBulkCopy.Close();//关闭连接                }            }            catch (Exception ex)            {                WriteErrorLog(ex.Message + &quot;数据库处理出错654行。SqlBulkCopyInsert&quot;);                throw (ex);            }        }        #endregion  <img  />

使用SqlBulkCopy将DataTable中的数据批量插入数据库中

<img    /><img    /><img  />
//BCP copy      SqlConnection conn = new SqlConnection();      conn.ConnectionString = &quot;server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS&quot;;      conn.Open();        SqlTransaction sqlbulkTransaction = conn.BeginTransaction();        //请在插入数据的同时检查约束,如果发生错误调用sqlbulkTransaction事务      SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction);          copy.DestinationTableName = &quot;T_SMS_SendInfo&quot;;      foreach (DataColumn dc in dataTable.Columns)      {          copy.ColumnMappings.Add(dc.ColumnName, dc.ColumnName);        }      try      {          copy.WriteToServer(dataTable);          sqlbulkTransaction.Commit();      }      catch (Exception ex)      {          sqlbulkTransaction.Rollback();          Console.WriteLine(ex.ToString());      }      finally      {          copy.Close();          conn.Close();      }  <img  />









在之前只知道SqlServer支持数据批量插入,殊不知道Oracle、SQLite和MySql也是支持的,不过Oracle需要使用Orace.DataAccess驱动,今天就贴出几种数据库的批量插入解决方法。


    首先说一下,IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider,此接口在前一篇文章中已经提到过了。
<img    /><img    /><img  />
  /// <summary>    /// 提供数据批量处理的方法。    /// </summary>    public interface IBatcherProvider : IProviderService    {        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        void Insert(DataTable dataTable, int batchSize = 10000);    } <img  />

View Code
一、SqlServer数据批量插入
    SqlServer的批量插入很简单,使用SqlBulkCopy就可以,以下是该类的实现:

<img    /><img    /><img  />
/// <summary>    /// 为 System.Data.SqlClient 提供的用于批量操作的方法。    /// </summary>    public sealed class MsSqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = (SqlConnection)ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    //给表名加上前后导符                    var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider>(), dataTable.TableName);                    using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)                        {                            DestinationTableName = tableName,                             BatchSize = batchSize                        })                    {                        //循环所有列,为bulk添加映射                        dataTable.EachColumn(c => bulk.ColumnMappings.Add(c.ColumnName, c.ColumnName), c => !c.AutoIncrement);                        bulk.WriteToServer(dataTable);                        bulk.Close();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }    }      SqlBulkCopy的ColumnMappings中列的名称受大小写敏感限制,因此在构造DataTable的时候应请注意列名要与表一致。     以上没有使用事务,使用事务在性能上会有一定的影响,如果要使用事务,可以设置SqlBulkCopyOptions.UseInternalTransaction。<img  />

View Code
二、Oracle数据批量插入
     System.Data.OracleClient不支持批量插入,因此只能使用Oracle.DataAccess组件来作为提供者。
<img    /><img    /><img  />
/// <summary>    /// Oracle.Data.Access 组件提供的用于批量操作的方法。    /// </summary>    public sealed class OracleAccessBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            //将一个DataTable的数据转换为数组的数组            var data = table.ToArray();            //设置ArrayBindCount属性            command.GetType().GetProperty(&quot;ArrayBindCount&quot;).SetValue(command, table.Rows.Count, null);            var syntax = database.Provider.GetService<ISyntaxProvider>();            for (var i = 0; i < table.Columns.Count; i++)            {                var column = table.Columns;                var parameter = database.Provider.DbProviderFactory.CreateParameter();                if (parameter == null)                {                    continue;                }                parameter.ParameterName = column.ColumnName;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = column.DataType.GetDbType();                parameter.Value = data;                if (names.Length > 0)                {                    names.Append(&quot;,&quot;);                    values.Append(&quot;,&quot;);                }                names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, column.ColumnName));                values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                command.Parameters.Add(parameter);            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
    以上最重要的一步,就是将DataTable转为数组的数组表示,即object[][],前数组的上标是列的个数,后数组是行的个数,因此循环Columns将后数组作为Parameter的值,也就是说,参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。

2020-11-4 12:09:01 评论

举报

三、SQLite数据批量插入
     SQLite的批量插入只需开启事务就可以了,这个具体的原理不得而知。
<img    /><img    /><img  />
public sealed class SQLiteBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                DbTransaction transcation = null;                try                {                    connection.TryOpen();                    transcation = connection.BeginTransaction();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        var flag = new AssertFlag();                        dataTable.EachRow(row =>                            {                                var first = flag.AssertTrue();                                ProcessCommandParameters(dataTable, command, row, first);                                command.ExecuteNonQuery();                            });                    }                    transcation.Commit();                }                catch (Exception exp)                {                    if (transcation != null)                    {                        transcation.Rollback();                    }                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        private void ProcessCommandParameters(DataTable dataTable, DbCommand command, DataRow row, bool first)        {            for (var c = 0; c < dataTable.Columns.Count; c++)            {                DbParameter parameter;                //首次创建参数,是为了使用缓存                if (first)                {                    parameter = ServiceContext.Database.Provider.DbProviderFactory.CreateParameter();                    parameter.ParameterName = dataTable.Columns[c].ColumnName;                    command.Parameters.Add(parameter);                }                else                {                    parameter = command.Parameters[c];                }                parameter.Value = row[c];            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DataTable table)        {            var syntax = database.Provider.GetService<ISyntaxProvider>();            var names = new StringBuilder();            var values = new StringBuilder();            var flag = new AssertFlag();            table.EachColumn(column =>                {                    if (!flag.AssertTrue())                    {                        names.Append(&quot;,&quot;);                        values.Append(&quot;,&quot;);                    }                    names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName));                    values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                });            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
   四、MySql数据批量插入
<img    /><img    /><img  />
/// <summary>    /// 为 MySql.Data 组件提供的用于批量操作的方法。    /// </summary>    public sealed class MySqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            var types = new List<DbType>();            var count = table.Columns.Count;            var syntax = database.Provider.GetService<ISyntaxProvider>();            table.EachColumn(c =>                {                    if (names.Length > 0)                    {                        names.Append(&quot;,&quot;);                    }                    names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, c.ColumnName));                    types.Add(c.DataType.GetDbType());                });            var i = 0;            foreach (DataRow row in table.Rows)            {                if (i > 0)                {                    values.Append(&quot;,&quot;);                }                values.Append(&quot;(&quot;);                for (var j = 0; j < count; j++)                {                    if (j > 0)                    {                        values.Append(&quot;, &quot;);                    }                    var isStrType = IsStringType(types[j]);                    var parameter = CreateParameter(database.Provider, isStrType, types[j], row[j], syntax.ParameterPrefix, i, j);                    if (parameter != null)                    {                        values.Append(parameter.ParameterName);                        command.Parameters.Add(parameter);                    }                    else if (isStrType)                    {                        values.AppendFormat(&quot;&#39;{0}&#39;&quot;, row[j]);                    }                    else                    {                        values.Append(row[j]);                    }                }                values.Append(&quot;)&quot;);                i++;            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES {2}&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }        /// <summary>        /// 判断是否为字符串类别。        /// </summary>        /// <param name=&quot;dbType&quot;></param>        /// <returns></returns>        private bool IsStringType(DbType dbType)        {            return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength;        }        /// <summary>        /// 创建参数。        /// </summary>        /// <param name=&quot;provider&quot;></param>        /// <param name=&quot;isStrType&quot;></param>        /// <param name=&quot;dbType&quot;></param>        /// <param name=&quot;value&quot;></param>        /// <param name=&quot;parPrefix&quot;></param>        /// <param name=&quot;row&quot;></param>        /// <param name=&quot;col&quot;></param>        /// <returns></returns>        private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col)        {            //如果生成全部的参数,则速度会很慢,因此,只有数据类型为字符串(包含&#39;号)和日期型时才添加参数            if ((isStrType && value.ToString().IndexOf(&#39;/&#39;&#39;) != -1) || dbType == DbType.DateTime)            {                var name = string.Format(&quot;{0}p_{1}_{2}&quot;, parPrefix, row, col);                var parameter = provider.DbProviderFactory.CreateParameter();                parameter.ParameterName = name;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = dbType;                parameter.Value = value;                return parameter;            }            return null;        }    }<img  />

View Code
    MySql的批量插入,是将值全部写在语句的values里,例如,insert batcher(id, name) values(1, &#39;1&#39;, 2, &#39;2&#39;, 3, &#39;3&#39;, ........ 10, &#39;10&#39;)。

     五、测试
     接下来写一个测试用例来看一下使用批量插入的效果。
<img    /><img    /><img  />
[Test]        public void TestBatchInsert()        {            Console.WriteLine(TimeWatcher.Watch(() =>                InvokeTest(database =>                    {                        var table = new DataTable(&quot;Batcher&quot;);                        table.Columns.Add(&quot;Id&quot;, typeof(int));                        table.Columns.Add(&quot;Name1&quot;, typeof(string));                        table.Columns.Add(&quot;Name2&quot;, typeof(string));                        table.Columns.Add(&quot;Name3&quot;, typeof(string));                        table.Columns.Add(&quot;Name4&quot;, typeof(string));                        //构造100000条数据                        for (var i = 0; i < 100000; i++)                        {                            table.Rows.Add(i, i.ToString(), i.ToString(), i.ToString(), i.ToString());                        }                        //获取 IBatcherProvider                        var batcher = database.Provider.GetService<IBatcherProvider>();                        if (batcher == null)                        {                            Console.WriteLine(&quot;不支持批量插入。&quot;);                        }                        else                        {                            batcher.Insert(table);                        }                        //输出batcher表的数据量                        var sql = new SqlCommand(&quot;SELECT COUNT(1) FROM Batcher&quot;);                        Console.WriteLine(&quot;当前共有 {0} 条数据&quot;, database.ExecuteScalar(sql));                    })));        } <img  />

View Code
  以下表中列出了四种数据库生成10万条数据各耗用的时间

数据库

耗用时间

MsSql00:00:02.9376300
Oracle00:00:01.5155959
SQLite00:00:01.6275634
MySql00:00:05.4166891
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
<img    /><img    /><img  />
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// <summary>        /// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// </summary>        /// <param name=&quot;strTableName&quot;>数据库中对应的表名</param>        /// <param name=&quot;dtData&quot;>数据集</param>        public void SqlBulkCopyInsert(string strTableName, DataTable dtData)        {            string ConStr = connectionString;// 数据库连接字符串              try            {                using (SqlBulkCopy sqlRevdBulkCopy = new SqlBulkCopy(ConStr))//引用SqlBulkCopy                {                    sqlRevdBulkCopy.DestinationTableName = strTableName;//数据库中对应的表名                      sqlRevdBulkCopy.NotifyAfter = dtData.Rows.Count;//有几行数据                      sqlRevdBulkCopy.WriteToServer(dtData);//数据导入数据库                      sqlRevdBulkCopy.Close();//关闭连接                }            }            catch (Exception ex)            {                WriteErrorLog(ex.Message + &quot;数据库处理出错654行。SqlBulkCopyInsert&quot;);                throw (ex);            }        }        #endregion  <img  />

使用SqlBulkCopy将DataTable中的数据批量插入数据库中

<img    /><img    /><img  />
//BCP copy      SqlConnection conn = new SqlConnection();      conn.ConnectionString = &quot;server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS&quot;;      conn.Open();        SqlTransaction sqlbulkTransaction = conn.BeginTransaction();        //请在插入数据的同时检查约束,如果发生错误调用sqlbulkTransaction事务      SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction);          copy.DestinationTableName = &quot;T_SMS_SendInfo&quot;;      foreach (DataColumn dc in dataTable.Columns)      {          copy.ColumnMappings.Add(dc.ColumnName, dc.ColumnName);        }      try      {          copy.WriteToServer(dataTable);          sqlbulkTransaction.Commit();      }      catch (Exception ex)      {          sqlbulkTransaction.Rollback();          Console.WriteLine(ex.ToString());      }      finally      {          copy.Close();          conn.Close();      }  <img  />







    首先说一下,IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider,此接口在前一篇文章中已经提到过了。
<img    /><img    /><img  />
  /// <summary>    /// 提供数据批量处理的方法。    /// </summary>    public interface IBatcherProvider : IProviderService    {        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        void Insert(DataTable dataTable, int batchSize = 10000);    } <img  />

View Code
一、SqlServer数据批量插入
    SqlServer的批量插入很简单,使用SqlBulkCopy就可以,以下是该类的实现:

<img    /><img    /><img  />
/// <summary>    /// 为 System.Data.SqlClient 提供的用于批量操作的方法。    /// </summary>    public sealed class MsSqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = (SqlConnection)ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    //给表名加上前后导符                    var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider>(), dataTable.TableName);                    using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)                        {                            DestinationTableName = tableName,                             BatchSize = batchSize                        })                    {                        //循环所有列,为bulk添加映射                        dataTable.EachColumn(c => bulk.ColumnMappings.Add(c.ColumnName, c.ColumnName), c => !c.AutoIncrement);                        bulk.WriteToServer(dataTable);                        bulk.Close();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }    }      SqlBulkCopy的ColumnMappings中列的名称受大小写敏感限制,因此在构造DataTable的时候应请注意列名要与表一致。     以上没有使用事务,使用事务在性能上会有一定的影响,如果要使用事务,可以设置SqlBulkCopyOptions.UseInternalTransaction。<img  />

View Code
二、Oracle数据批量插入
     System.Data.OracleClient不支持批量插入,因此只能使用Oracle.DataAccess组件来作为提供者。
<img    /><img    /><img  />
/// <summary>    /// Oracle.Data.Access 组件提供的用于批量操作的方法。    /// </summary>    public sealed class OracleAccessBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            //将一个DataTable的数据转换为数组的数组            var data = table.ToArray();            //设置ArrayBindCount属性            command.GetType().GetProperty(&quot;ArrayBindCount&quot;).SetValue(command, table.Rows.Count, null);            var syntax = database.Provider.GetService<ISyntaxProvider>();            for (var i = 0; i < table.Columns.Count; i++)            {                var column = table.Columns;                var parameter = database.Provider.DbProviderFactory.CreateParameter();                if (parameter == null)                {                    continue;                }                parameter.ParameterName = column.ColumnName;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = column.DataType.GetDbType();                parameter.Value = data;                if (names.Length > 0)                {                    names.Append(&quot;,&quot;);                    values.Append(&quot;,&quot;);                }                names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, column.ColumnName));                values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                command.Parameters.Add(parameter);            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
    以上最重要的一步,就是将DataTable转为数组的数组表示,即object[][],前数组的上标是列的个数,后数组是行的个数,因此循环Columns将后数组作为Parameter的值,也就是说,参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。
三、SQLite数据批量插入
     SQLite的批量插入只需开启事务就可以了,这个具体的原理不得而知。
<img    /><img    /><img  />
public sealed class SQLiteBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                DbTransaction transcation = null;                try                {                    connection.TryOpen();                    transcation = connection.BeginTransaction();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        var flag = new AssertFlag();                        dataTable.EachRow(row =>                            {                                var first = flag.AssertTrue();                                ProcessCommandParameters(dataTable, command, row, first);                                command.ExecuteNonQuery();                            });                    }                    transcation.Commit();                }                catch (Exception exp)                {                    if (transcation != null)                    {                        transcation.Rollback();                    }                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        private void ProcessCommandParameters(DataTable dataTable, DbCommand command, DataRow row, bool first)        {            for (var c = 0; c < dataTable.Columns.Count; c++)            {                DbParameter parameter;                //首次创建参数,是为了使用缓存                if (first)                {                    parameter = ServiceContext.Database.Provider.DbProviderFactory.CreateParameter();                    parameter.ParameterName = dataTable.Columns[c].ColumnName;                    command.Parameters.Add(parameter);                }                else                {                    parameter = command.Parameters[c];                }                parameter.Value = row[c];            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DataTable table)        {            var syntax = database.Provider.GetService<ISyntaxProvider>();            var names = new StringBuilder();            var values = new StringBuilder();            var flag = new AssertFlag();            table.EachColumn(column =>                {                    if (!flag.AssertTrue())                    {                        names.Append(&quot;,&quot;);                        values.Append(&quot;,&quot;);                    }                    names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName));                    values.AppendFormat(&quot;{0}{1}&quot;, syntax.ParameterPrefix, column.ColumnName);                });            return string.Format(&quot;INSERT INTO {0}({1}) VALUES ({2})&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }    }<img  />

View Code
   四、MySql数据批量插入
<img    /><img    /><img  />
/// <summary>    /// 为 MySql.Data 组件提供的用于批量操作的方法。    /// </summary>    public sealed class MySqlBatcher : IBatcherProvider    {        /// <summary>        /// 获取或设置提供者服务的上下文。        /// </summary>        public ServiceContext ServiceContext { get; set; }        /// <summary>        /// 将 <see cref=&quot;DataTable&quot;/> 的数据批量插入到数据库中。        /// </summary>        /// <param name=&quot;dataTable&quot;>要批量插入的 <see cref=&quot;DataTable&quot;/>。</param>        /// <param name=&quot;batchSize&quot;>每批次写入的数据量。</param>        public void Insert(DataTable dataTable, int batchSize = 10000)        {            Checker.ArgumentNull(dataTable, &quot;dataTable&quot;);            if (dataTable.Rows.Count == 0)            {                return;            }            using (var connection = ServiceContext.Database.CreateConnection())            {                try                {                    connection.TryOpen();                    using (var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand())                    {                        if (command == null)                        {                            throw new BatcherException(new ArgumentException(&quot;command&quot;));                        }                        command.Connection = connection;                        command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable);                        if (command.CommandText == string.Empty)                        {                            return;                        }                        command.ExecuteNonQuery();                    }                }                catch (Exception exp)                {                    throw new BatcherException(exp);                }                finally                {                    connection.TryClose();                }            }        }        /// <summary>        /// 生成插入数据的sql语句。        /// </summary>        /// <param name=&quot;database&quot;></param>        /// <param name=&quot;command&quot;></param>        /// <param name=&quot;table&quot;></param>        /// <returns></returns>        private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table)        {            var names = new StringBuilder();            var values = new StringBuilder();            var types = new List<DbType>();            var count = table.Columns.Count;            var syntax = database.Provider.GetService<ISyntaxProvider>();            table.EachColumn(c =>                {                    if (names.Length > 0)                    {                        names.Append(&quot;,&quot;);                    }                    names.AppendFormat(&quot;{0}&quot;, DbUtility.FormatByQuote(syntax, c.ColumnName));                    types.Add(c.DataType.GetDbType());                });            var i = 0;            foreach (DataRow row in table.Rows)            {                if (i > 0)                {                    values.Append(&quot;,&quot;);                }                values.Append(&quot;(&quot;);                for (var j = 0; j < count; j++)                {                    if (j > 0)                    {                        values.Append(&quot;, &quot;);                    }                    var isStrType = IsStringType(types[j]);                    var parameter = CreateParameter(database.Provider, isStrType, types[j], row[j], syntax.ParameterPrefix, i, j);                    if (parameter != null)                    {                        values.Append(parameter.ParameterName);                        command.Parameters.Add(parameter);                    }                    else if (isStrType)                    {                        values.AppendFormat(&quot;&#39;{0}&#39;&quot;, row[j]);                    }                    else                    {                        values.Append(row[j]);                    }                }                values.Append(&quot;)&quot;);                i++;            }            return string.Format(&quot;INSERT INTO {0}({1}) VALUES {2}&quot;, DbUtility.FormatByQuote(syntax, table.TableName), names, values);        }        /// <summary>        /// 判断是否为字符串类别。        /// </summary>        /// <param name=&quot;dbType&quot;></param>        /// <returns></returns>        private bool IsStringType(DbType dbType)        {            return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength;        }        /// <summary>        /// 创建参数。        /// </summary>        /// <param name=&quot;provider&quot;></param>        /// <param name=&quot;isStrType&quot;></param>        /// <param name=&quot;dbType&quot;></param>        /// <param name=&quot;value&quot;></param>        /// <param name=&quot;parPrefix&quot;></param>        /// <param name=&quot;row&quot;></param>        /// <param name=&quot;col&quot;></param>        /// <returns></returns>        private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col)        {            //如果生成全部的参数,则速度会很慢,因此,只有数据类型为字符串(包含&#39;号)和日期型时才添加参数            if ((isStrType && value.ToString().IndexOf(&#39;/&#39;&#39;) != -1) || dbType == DbType.DateTime)            {                var name = string.Format(&quot;{0}p_{1}_{2}&quot;, parPrefix, row, col);                var parameter = provider.DbProviderFactory.CreateParameter();                parameter.ParameterName = name;                parameter.Direction = ParameterDirection.Input;                parameter.DbType = dbType;                parameter.Value = value;                return parameter;            }            return null;        }    }<img  />

View Code
    MySql的批量插入,是将值全部写在语句的values里,例如,insert batcher(id, name) values(1, &#39;1&#39;, 2, &#39;2&#39;, 3, &#39;3&#39;, ........ 10, &#39;10&#39;)。

     五、测试
     接下来写一个测试用例来看一下使用批量插入的效果。
<img    /><img    /><img  />
[Test]        public void TestBatchInsert()        {            Console.WriteLine(TimeWatcher.Watch(() =>                InvokeTest(database =>                    {                        var table = new DataTable(&quot;Batcher&quot;);                        table.Columns.Add(&quot;Id&quot;, typeof(int));                        table.Columns.Add(&quot;Name1&quot;, typeof(string));                        table.Columns.Add(&quot;Name2&quot;, typeof(string));                        table.Columns.Add(&quot;Name3&quot;, typeof(string));                        table.Columns.Add(&quot;Name4&quot;, typeof(string));                        //构造100000条数据                        for (var i = 0; i < 100000; i++)                        {                            table.Rows.Add(i, i.ToString(), i.ToString(), i.ToString(), i.ToString());                        }                        //获取 IBatcherProvider                        var batcher = database.Provider.GetService<IBatcherProvider>();                        if (batcher == null)                        {                            Console.WriteLine(&quot;不支持批量插入。&quot;);                        }                        else                        {                            batcher.Insert(table);                        }                        //输出batcher表的数据量                        var sql = new SqlCommand(&quot;SELECT COUNT(1) FROM Batcher&quot;);                        Console.WriteLine(&quot;当前共有 {0} 条数据&quot;, database.ExecuteScalar(sql));                    })));        } <img  />

View Code
  以下表中列出了四种数据库生成10万条数据各耗用的时间

数据库

耗用时间

MsSql00:00:02.9376300
Oracle00:00:01.5155959
SQLite00:00:01.6275634
MySql00:00:05.4166891
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
<img    /><img    /><img  />
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// <summary>        /// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中        /// </summary>        /// <param name=&quot;strTableName&quot;>数据库中对应的表名</param>        /// <param name=&quot;dtData&quot;>数据集</param>        public void SqlBulkCopyInsert(string strTableName, DataTable dtData)        {            string ConStr = connectionString;// 数据库连接字符串              try            {                using (SqlBulkCopy sqlRevdBulkCopy = new SqlBulkCopy(ConStr))//引用SqlBulkCopy                {                    sqlRevdBulkCopy.DestinationTableName = strTableName;//数据库中对应的表名                      sqlRevdBulkCopy.NotifyAfter = dtData.Rows.Count;//有几行数据                      sqlRevdBulkCopy.WriteToServer(dtData);//数据导入数据库                      sqlRevdBulkCopy.Close();//关闭连接                }            }            catch (Exception ex)            {                WriteErrorLog(ex.Message + &quot;数据库处理出错654行。SqlBulkCopyInsert&quot;);                throw (ex);            }        }        #endregion  <img  />

使用SqlBulkCopy将DataTable中的数据批量插入数据库中

<img    /><img    /><img  />
//BCP copy      SqlConnection conn = new SqlConnection();      conn.ConnectionString = &quot;server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS&quot;;      conn.Open();        SqlTransaction sqlbulkTransaction = conn.BeginTransaction();        //请在插入数据的同时检查约束,如果发生错误调用sqlbulkTransaction事务      SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction);          copy.DestinationTableName = &quot;T_SMS_SendInfo&quot;;      foreach (DataColumn dc in dataTable.Columns)      {          copy.ColumnMappings.Add(dc.ColumnName, dc.ColumnName);        }      try      {          copy.WriteToServer(dataTable);          sqlbulkTransaction.Commit();      }      catch (Exception ex)      {          sqlbulkTransaction.Rollback();          Console.WriteLine(ex.ToString());      }      finally      {          copy.Close();          conn.Close();      }  <img  />
2020-11-4 12:09:07 评论

举报

撰写答案

你正在撰写答案

如果你是对答案或其他答案精选点评或询问,请使用“评论”功能。

高级模式
您需要登录后才可以回帖 登录 | 注册

声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表电子发烧友网立场。文章及其配图仅供工程师学习之用,如有内容图片侵权或者其他问题,请联系本站作侵删。 侵权投诉
我要提问
关闭

站长推荐 上一条 /9 下一条

快速回复 返回顶部 返回列表