三、SQLite数据批量插入
SQLite的批量插入只需开启事务就可以了，这个具体的原理不得而知。
public sealed class SQLiteBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); 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("command")); } 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]; } } /// /// 生成插入数据的sql语句。 /// /// /// /// private string GenerateInserSql(IDatabase database, DataTable table) { var syntax = database.Provider.GetService(); var names = new StringBuilder(); var values = new StringBuilder(); var flag = new AssertFlag(); table.EachColumn(column => { if (!flag.AssertTrue()) { names.Append(","); values.Append(","); } names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName)); values.AppendFormat("{0}{1}", syntax.ParameterPrefix, column.ColumnName); }); return string.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }
View Code
四、MySql数据批量插入
/// /// 为 MySql.Data 组件提供的用于批量操作的方法。 /// public sealed class MySqlBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); 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("command")); } 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(); } } } /// /// 生成插入数据的sql语句。 /// /// /// /// /// private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); var types = new List(); var count = table.Columns.Count; var syntax = database.Provider.GetService(); table.EachColumn(c => { if (names.Length > 0) { names.Append(","); } names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, c.ColumnName)); types.Add(c.DataType.GetDbType()); }); var i = 0; foreach (DataRow row in table.Rows) { if (i > 0) { values.Append(","); } values.Append("("); for (var j = 0; j < count; j++) { if (j > 0) { values.Append(", "); } 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("'{0}'", row[j]); } else { values.Append(row[j]); } } values.Append(")"); i++; } return string.Format("INSERT INTO {0}({1}) VALUES {2}", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } /// /// 判断是否为字符串类别。 /// /// /// private bool IsStringType(DbType dbType) { return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength; } /// /// 创建参数。 /// /// /// /// /// /// /// /// /// private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col) { //如果生成全部的参数，则速度会很慢，因此，只有数据类型为字符串(包含'号)和日期型时才添加参数 if ((isStrType && value.ToString().IndexOf('/'') != -1) || dbType == DbType.DateTime) { var name = string.Format("{0}p_{1}_{2}", 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; } }
View Code
MySql的批量插入，是将值全部写在语句的values里，例如，insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。
五、测试
接下来写一个测试用例来看一下使用批量插入的效果。
[Test] public void TestBatchInsert() { Console.WriteLine(TimeWatcher.Watch(() => InvokeTest(database => { var table = new DataTable("Batcher"); table.Columns.Add("Id", typeof(int)); table.Columns.Add("Name1", typeof(string)); table.Columns.Add("Name2", typeof(string)); table.Columns.Add("Name3", typeof(string)); table.Columns.Add("Name4", 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(); if (batcher == null) { Console.WriteLine("不支持批量插入。"); } else { batcher.Insert(table); } //输出batcher表的数据量 var sql = new SqlCommand("SELECT COUNT(1) FROM Batcher"); Console.WriteLine("当前共有 {0} 条数据", database.ExecuteScalar(sql)); }))); }
View Code
以下表中列出了四种数据库生成10万条数据各耗用的时间
数据库
| 耗用时间
| | MsSql | 00:00:02.9376300 | | Oracle | 00:00:01.5155959 | | SQLite | 00:00:01.6275634 | | MySql | 00:00:05.4166891 |
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中 /// /// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中 /// /// 数据库中对应的表名 /// 数据集 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 + "数据库处理出错654行。SqlBulkCopyInsert"); throw (ex); } } #endregion
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
//BCP copy SqlConnection conn = new SqlConnection(); conn.ConnectionString = "server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS"; conn.Open(); SqlTransaction sqlbulkTransaction = conn.BeginTransaction(); //请在插入数据的同时检查约束，如果发生错误调用sqlbulkTransaction事务 SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction); copy.DestinationTableName = "T_SMS_SendInfo"; 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(); }
首先说一下，IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider，此接口在前一篇文章中已经提到过了。
/// /// 提供数据批量处理的方法。 /// public interface IBatcherProvider : IProviderService { /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 void Insert(DataTable dataTable, int batchSize = 10000); }
View Code
一、SqlServer数据批量插入
SqlServer的批量插入很简单，使用SqlBulkCopy就可以，以下是该类的实现：
/// /// 为 System.Data.SqlClient 提供的用于批量操作的方法。 /// public sealed class MsSqlBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); if (dataTable.Rows.Count == 0) { return; } using (var connection = (SqlConnection)ServiceContext.Database.CreateConnection()) { try { connection.TryOpen(); //给表名加上前后导符 var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService(), 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。
View Code
二、Oracle数据批量插入
System.Data.OracleClient不支持批量插入，因此只能使用Oracle.DataAccess组件来作为提供者。
/// /// Oracle.Data.Access 组件提供的用于批量操作的方法。 /// public sealed class OracleAccessBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); 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("command")); } command.Connection = connection; command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable); command.ExecuteNonQuery(); } } catch (Exception exp) { throw new BatcherException(exp); } finally { connection.TryClose(); } } } /// /// 生成插入数据的sql语句。 /// /// /// /// /// 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("ArrayBindCount").SetValue(command, table.Rows.Count, null); var syntax = database.Provider.GetService(); 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(","); values.Append(","); } names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, column.ColumnName)); values.AppendFormat("{0}{1}", syntax.ParameterPrefix, column.ColumnName); command.Parameters.Add(parameter); } return string.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }
View Code
以上最重要的一步，就是将DataTable转为数组的数组表示，即object[][]，前数组的上标是列的个数，后数组是行的个数，因此循环Columns将后数组作为Parameter的值，也就是说，参数的值是一个数组。而insert语句与一般的插入语句没有什么不一样。
三、SQLite数据批量插入
SQLite的批量插入只需开启事务就可以了，这个具体的原理不得而知。
public sealed class SQLiteBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); 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("command")); } 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]; } } /// /// 生成插入数据的sql语句。 /// /// /// /// private string GenerateInserSql(IDatabase database, DataTable table) { var syntax = database.Provider.GetService(); var names = new StringBuilder(); var values = new StringBuilder(); var flag = new AssertFlag(); table.EachColumn(column => { if (!flag.AssertTrue()) { names.Append(","); values.Append(","); } names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName)); values.AppendFormat("{0}{1}", syntax.ParameterPrefix, column.ColumnName); }); return string.Format("INSERT INTO {0}({1}) VALUES ({2})", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }
View Code
四、MySql数据批量插入
/// /// 为 MySql.Data 组件提供的用于批量操作的方法。 /// public sealed class MySqlBatcher : IBatcherProvider { /// /// 获取或设置提供者服务的上下文。 /// public ServiceContext ServiceContext { get; set; } /// /// 将 的数据批量插入到数据库中。 /// /// 要批量插入的 。 /// 每批次写入的数据量。 public void Insert(DataTable dataTable, int batchSize = 10000) { Checker.ArgumentNull(dataTable, "dataTable"); 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("command")); } 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(); } } } /// /// 生成插入数据的sql语句。 /// /// /// /// /// private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); var types = new List(); var count = table.Columns.Count; var syntax = database.Provider.GetService(); table.EachColumn(c => { if (names.Length > 0) { names.Append(","); } names.AppendFormat("{0}", DbUtility.FormatByQuote(syntax, c.ColumnName)); types.Add(c.DataType.GetDbType()); }); var i = 0; foreach (DataRow row in table.Rows) { if (i > 0) { values.Append(","); } values.Append("("); for (var j = 0; j < count; j++) { if (j > 0) { values.Append(", "); } 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("'{0}'", row[j]); } else { values.Append(row[j]); } } values.Append(")"); i++; } return string.Format("INSERT INTO {0}({1}) VALUES {2}", DbUtility.FormatByQuote(syntax, table.TableName), names, values); } /// /// 判断是否为字符串类别。 /// /// /// private bool IsStringType(DbType dbType) { return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength; } /// /// 创建参数。 /// /// /// /// /// /// /// /// /// private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col) { //如果生成全部的参数，则速度会很慢，因此，只有数据类型为字符串(包含'号)和日期型时才添加参数 if ((isStrType && value.ToString().IndexOf('/'') != -1) || dbType == DbType.DateTime) { var name = string.Format("{0}p_{1}_{2}", 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; } }
View Code
MySql的批量插入，是将值全部写在语句的values里，例如，insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。
五、测试
接下来写一个测试用例来看一下使用批量插入的效果。
[Test] public void TestBatchInsert() { Console.WriteLine(TimeWatcher.Watch(() => InvokeTest(database => { var table = new DataTable("Batcher"); table.Columns.Add("Id", typeof(int)); table.Columns.Add("Name1", typeof(string)); table.Columns.Add("Name2", typeof(string)); table.Columns.Add("Name3", typeof(string)); table.Columns.Add("Name4", 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(); if (batcher == null) { Console.WriteLine("不支持批量插入。"); } else { batcher.Insert(table); } //输出batcher表的数据量 var sql = new SqlCommand("SELECT COUNT(1) FROM Batcher"); Console.WriteLine("当前共有 {0} 条数据", database.ExecuteScalar(sql)); }))); }
View Code
以下表中列出了四种数据库生成10万条数据各耗用的时间
数据库
| 耗用时间
| | MsSql | 00:00:02.9376300 | | Oracle | 00:00:01.5155959 | | SQLite | 00:00:01.6275634 | | MySql | 00:00:05.4166891 |
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
#region 使用SqlBulkCopy将DataTable中的数据批量插入数据库中 /// /// 使用SqlBulkCopy将DataTable中的数据批量插入数据库中 /// /// 数据库中对应的表名 /// 数据集 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 + "数据库处理出错654行。SqlBulkCopyInsert"); throw (ex); } } #endregion
使用SqlBulkCopy将DataTable中的数据批量插入数据库中
//BCP copy SqlConnection conn = new SqlConnection(); conn.ConnectionString = "server=.;uid=dmkj_hpc;pwd=#$wlh*&1110h%c;database=DMKJ_SMS"; conn.Open(); SqlTransaction sqlbulkTransaction = conn.BeginTransaction(); //请在插入数据的同时检查约束，如果发生错误调用sqlbulkTransaction事务 SqlBulkCopy copy = new SqlBulkCopy(conn, SqlBulkCopyOptions.CheckConstraints, sqlbulkTransaction); copy.DestinationTableName = "T_SMS_SendInfo"; 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(); }
|
|
|
|
2020-11-4 12:09:07
评论
举报
|
|
|
|
0
/6 
工商网监
湘ICP备2023018690号
1727
淘帖
2875
