Parallel.Invoke()、TransactionScope() 和 SqlBulkCopy

Question

Parallel.Invoke() 中有多个需要在事务内部运行的方法。这些方法都调用SqlBulkCopy 的实例。用例是“all-or-none”，因此如果一个方法失败，则不会提交任何内容。当我在父事务上调用 Complete() 方法时，我得到了 TransactionAbortedException ({"Transaction Timeout"})。

这是父事务：

using (var ts = new TransactionScope())
{
    var saveClone = Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete);
    var saveErrorsClone = Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete);
    var saveADClone = Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete);
    var saveEnrollmentsClone = Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete);
    Parallel.Invoke(_options, () =>
        {
            Save(data, saveClone);
        },
        () =>
        {
            SaveErrors(saveErrorsClone);
        },
        () =>
        {
            SaveEnrollments(data, saveEnrollmentsClone);
        });
ts.Complete();  
}//***** GET THE EXCEPTION HERE *****

这是一个使用 SqlBulkCopy 的依赖事务（它们都是相同的结构）。我正在传入父母并将其分配给孩子的TransactionScope

private void Save(IDictionary<string, string> data, Transaction transaction)
{
    var dTs = (DependentTransaction)transaction;
    if (transaction.TransactionInformation.Status != TransactionStatus.Aborted)
    {
        using (var ts = new TransactionScope(dTs))
        {
            _walmartData.Save(data);
            Debug.WriteLine("Completed Processing XML - {0}", _stopWatch.Elapsed);
            ts.Complete();
        }
    }
    else
    {
        Debug.WriteLine("Save Not Executed - Transaction Aborted - {0}", _stopWatch.Elapsed);    
        dTs.Complete();
    }
    dTs.Complete();
}

编辑（添加了我的SqlBulkCopy 方法...注意事务参数为空）

private void SqlBulkCopy(DataTable dt, SqlBulkCopyColumnMappingCollection mappings)
{
    try
    {
        using (var sbc = new SqlBulkCopy(_conn, SqlBulkCopyOptions.TableLock, null))
        {
            sbc.BatchSize = 100;
            sbc.BulkCopyTimeout = 0;
            sbc.DestinationTableName = dt.TableName;
            foreach (SqlBulkCopyColumnMapping mapping in mappings)
            {
                sbc.ColumnMappings.Add(mapping);
            }
            sbc.WriteToServer(dt);
        }
    }
    catch (Exception)
    {
        throw;
    }
}

除了修复错误之外，我还愿意接受替代方案。谢谢。

Answer 1

您正在使用您选择的DependentCloneOption.BlockCommitUntilComplete 创建一种死锁形式。

Parallel.Invoke 阻塞调用线程，直到其所有处理完成。试图由Parallel.Invoke 完成的作业在等待父事务完成时都处于阻塞状态（由于 DependentCloneOption）。所以两个人在互相等待……僵局。父事务最终超时并从阻塞中释放依赖事务，从而解除对调用线程的阻塞。

你能用DependentCloneOption.RollbackIfNotComplete吗？

Answer 2

http://msdn.microsoft.com/en-us/library/system.transactions.transactionscope.complete.aspx 表示TransactionScope.Complete 只提交它包含的事务，如果它是创建它的那个。由于您是从现有事务创建范围，因此我相信您需要在对范围调用完成之前提交事务。

来自 MSDN：

资源管理器之间的实际提交工作发生在 如果 TransactionScope 对象创建了 End Using 语句 交易。如果它没有创建事务，则发生提交 每当 CommittableTransaction 的所有者调用 Commit 目的。此时事务管理器调用资源 经理并通知他们提交或回滚，基于 是否在 TransactionScope 对象上调用了此方法

.

Answer 3

经过大量痛苦、研究和缺乏有效答案后，我必须相信我在问题中描述的堆栈是不可能的。我相信，痛点在 TransactionScope 和 SqlBulkCopy 之间。为了未来的观众，我把这个答案放在这里。如果有人可以证明可以做到，我很乐意将其作为答案删除。

Answer 4

我相信你如何创建你的_conn-instance 很重要，如果你创建它并在你的TransactionScope-instance 中打开它，任何与SqlBulkCopy 相关的问题都应该得到解决。

看看Can I use SqlBulkCopy inside Transaction 和Is it possible to use System.Transactions.TransactionScope with SqlBulkCopy? 看看它是否对你有帮助。

void MyMainMethod()
{
 using (var ts = new TransactionScope())
 {
  Parallell.InvokeOrWhatNotOrWhatEver(() => DoStuff());
 }
}

void DoStuff() 
{
 using (var sqlCon = new SqlConnection(conStr))
 {
  sqlCon.Open(); // ensure to open it before SqlBulkCopy can open it in another transactionscope.
  using (var bulk = new SqlBulkCopy(sqlCon))
  {
    // Do you stuff
    bulk.WriteToServer...
  }      

  ts.Complete(); // finish the transaction, ie commit
 }
}

简而言之：

1. 创建事务范围
2. 创建sql-connection并在事务范围下打开
3. 通过上面创建的连接创建和使用SqlBulkCopy-instance
4. 致电transaction.Complete()
5. 扔掉所有东西 :-)