交易范围和Task.When全部执行
Transaction Scope and Task.When all execution
意图:通过事务提交/回滚支持加速Sql插入操作
- 拥有大量数据(约 90 万个实体)
- 想将数据并行写入 SQL table,将按连接拆分数据(例如 3 个活动连接,因此每个连接 300 k)
- 希望在出现问题时回滚,因此尝试使用事务范围 (nested-TransactionScopeOption.Required)
问题:
看起来事务范围不支持这种类型的 Task.WhenAll 因此,如果在执行期间出现错误,中间写入将不会回滚
问: 所以我想知道它是否可以变通,或者它只是不适合利用事务范围的方式?
public void Upsert(IEnumerable<IEnumerable<Item>> splitPerConnection, DateTime endDate)
{
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
{
try
{
UpdateEndDate(endDate);
var insertTasks = splitPerConnection.Select(ch => Task.Run(() => Insert(ch)));
Task.WhenAll(insertTasks).GetAwaiter().GetResult();
scope.Complete();
}
catch (Exception ex)
{
throw ex;
}
}
}
private int Insert(IEnumerable<Item> items)
{
int affectedRows;
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
using (var sqlConnection = new SqlConnection(DockerMsSqlServerConnectionString))
{
sqlConnection.Open();
affectedRows = sqlConnection.Execute(ItemsQueries.Insert, items, commandTimeout: 0);
scope.Complete();
}
return affectedRows;
}
private int UpdateEndDate(DateTime endDate)
{
int affectedRows;
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
using (var sqlConnection = new SqlConnection(DockerMsSqlServerConnectionString))
{
sqlConnection.Open();
affectedRows = sqlConnection.Execute(ItemsQueries.UpdateEndDate, new { EndDate = endDate }, commandTimeout: 0);
scope.Complete();
}
return affectedRows;
}
您可以尝试利用 SqlBulkCopy,而不是将项目拆分到不同的连接对象。
Insert 2 million rows into SQL Server quickly
您可以通过修改IEnumerable<Item>
来创建数据集。所有并行操作都可以在此处应用,以将 IEnumerable<Item>
转换为数据集。创建数据集后,SqlBulkCopy 将帮助您。您可以创建3-4组数据集并执行插入查询3-4次。
在这种方法中,您可以维护一个数据库连接,这也有助于遵循数据库事务的 ACID 合规性。
意图:通过事务提交/回滚支持加速Sql插入操作
- 拥有大量数据(约 90 万个实体)
- 想将数据并行写入 SQL table,将按连接拆分数据(例如 3 个活动连接,因此每个连接 300 k)
- 希望在出现问题时回滚,因此尝试使用事务范围 (nested-TransactionScopeOption.Required)
问题:
看起来事务范围不支持这种类型的 Task.WhenAll 因此,如果在执行期间出现错误,中间写入将不会回滚
问: 所以我想知道它是否可以变通,或者它只是不适合利用事务范围的方式?
public void Upsert(IEnumerable<IEnumerable<Item>> splitPerConnection, DateTime endDate)
{
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
{
try
{
UpdateEndDate(endDate);
var insertTasks = splitPerConnection.Select(ch => Task.Run(() => Insert(ch)));
Task.WhenAll(insertTasks).GetAwaiter().GetResult();
scope.Complete();
}
catch (Exception ex)
{
throw ex;
}
}
}
private int Insert(IEnumerable<Item> items)
{
int affectedRows;
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
using (var sqlConnection = new SqlConnection(DockerMsSqlServerConnectionString))
{
sqlConnection.Open();
affectedRows = sqlConnection.Execute(ItemsQueries.Insert, items, commandTimeout: 0);
scope.Complete();
}
return affectedRows;
}
private int UpdateEndDate(DateTime endDate)
{
int affectedRows;
using (var scope = _transactionScopeFactory.Create(TransactionScopeOption.Required))
using (var sqlConnection = new SqlConnection(DockerMsSqlServerConnectionString))
{
sqlConnection.Open();
affectedRows = sqlConnection.Execute(ItemsQueries.UpdateEndDate, new { EndDate = endDate }, commandTimeout: 0);
scope.Complete();
}
return affectedRows;
}
您可以尝试利用 SqlBulkCopy,而不是将项目拆分到不同的连接对象。
Insert 2 million rows into SQL Server quickly
您可以通过修改IEnumerable<Item>
来创建数据集。所有并行操作都可以在此处应用,以将 IEnumerable<Item>
转换为数据集。创建数据集后,SqlBulkCopy 将帮助您。您可以创建3-4组数据集并执行插入查询3-4次。
在这种方法中,您可以维护一个数据库连接,这也有助于遵循数据库事务的 ACID 合规性。