c#幾種資料庫的大數據批次插入（SqlServer、Oracle、SQLite和MySql）
- 高洛峰原創
- 2017-01-13 15:40:192480瀏覽
在之前只知道SqlServer支援資料批次插入，殊不知道Oracle、SQLite和MySql也是支援的，不過Oracle需要使用Orace.DataAccess驅動，今天就貼出幾種資料庫的批次插入解決方法。
首先說一下，IProvider裡有一個用於實現批次插入的插件服務介面IBatcherProvider，此介面在前一篇文章中已經提到過了。
/// <summary>
/// 提供数据批量处理的方法。
/// </summary>
public interface IBatcherProvider : IProviderService
{
/// <summary>
/// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次写入的数据量。</param>
void Insert(DataTable dataTable, int batchSize = 10000);
}一、SqlServer資料批次插入
SqlServer的批次插入很簡單，使用SqlBulkCopy就可以，以下是該類別的實作：
/// <summary>
/// 为 System.Data.SqlClient 提供的用于批量操作的方法。
/// </summary>
public sealed class MsSqlBatcher : IBatcherProvider
{
/// <summary>
/// 获取或设置提供者服务的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次写入的数据量。</param>
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<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();
}
}
}
}以上沒有使用事務，使用事務在效能上會有一定的影響，如果
/// <summary>
/// Oracle.Data.Access 组件提供的用于批量操作的方法。
/// </summary>
public sealed class OracleAccessBatcher : IBatcherProvider
{
/// <summary>
/// 获取或设置提供者服务的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次写入的数据量。</param>
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();
}
}
}
/// <summary>
/// 生成插入数据的sql语句。
/// </summary>
/// <param name="database"></param>
/// <param name="command"></param>
/// <param name="table"></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("ArrayBindCount").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[i];
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[i];
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);
}
}以上沒有使用事務，使用事務在效能上會有一定的影響，如果
public sealed class SQLiteBatcher : IBatcherProvider
{
/// <summary>
/// 获取或设置提供者服务的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次写入的数据量。</param>
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];
}
}
/// <summary>
/// 生成插入数据的sql语句。
/// </summary>
/// <param name="database"></param>
/// <param name="table"></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(",");
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);
}
}以上沒有使用事務，使用事務在效能上會有一定的影響要使用事務，可以設定SqlBulkCopyOptions.UseInternalTransaction。
二、Oracle資料批次插入
System.Data.OracleClient不支援批次插入，因此只能使用Oracle.DataAccess元件來作為提供者。
/// <summary>
/// 为 MySql.Data 组件提供的用于批量操作的方法。
/// </summary>
public sealed class MySqlBatcher : IBatcherProvider
{
/// <summary>
/// 获取或设置提供者服务的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 将 <see cref="DataTable"/> 的数据批量插入到数据库中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次写入的数据量。</param>
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();
}
}
}
/// <summary>
/// 生成插入数据的sql语句。
/// </summary>
/// <param name="database"></param>
/// <param name="command"></param>
/// <param name="table"></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(",");
}
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);
}
/// <summary>
/// 判断是否为字符串类别。
/// </summary>
/// <param name="dbType"></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="provider"></param>
/// <param name="isStrType"></param>
/// <param name="dbType"></param>
/// <param name="value"></param>
/// <param name="parPrefix"></param>
/// <param name="row"></param>
/// <param name="col"></param>
/// <returns></returns>
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;
}
}以上最重要的一步，就是將DataTable轉為數組的數組表示，即object[][]，前數組的上標是列的個數，後數組是行的個數，因此循環Columns將後數組作為Parameter的值，也就是說，參數的值是一個陣列。而insert語句與一般的插入語句沒有什麼不一樣。
三、SQLite資料批次插入
SQLite的批次插入只需開啟事務就可以了，這個具體的原理不得而知。
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<IBatcherProvider>();
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));
})));
}四、MySql資料批次插入
rrreeeMySql的批次插入，是將數值全部寫在語句的values裡，例如，insert batcher(id, name) values(1, '1', 2, '2' , 3, '3', ........ 10, '10')。
五、測試
接下來寫一個測試案例來看一下使用批次插入的效果。 
rrreee
陳述：
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
上一篇：asp.net驗證碼的簡單製作下一篇：asp.net驗證碼的簡單製作