为何SqlBulkCopy.WriteToServerAsync不遵循await关键字？附问题排查

这个问题的核心原因其实很隐蔽——你虽然在WriteToServerAsync前面加了await，但调用异步委托的地方根本没有等待异步操作完成，直接就清空了DataTable，导致后台还在运行的批量插入操作访问到了已经被移除的行。

问题根源分析 #

看你的ReadXMLInBatches方法，它是一个同步方法，但接收的processBatch是返回Task的异步委托。当批次数据满了之后，你执行了：

processBatch(table);
table.Rows.Clear();

这里的processBatch(table)只是发起了异步插入操作，并没有等待它完成就立刻清空了DataTable。而WriteToServerAsync内部是异步处理DataTable的行数据的，这时候后台线程还在读取DataTable的行，结果行被清空了，自然就抛出RowNotInTableException。

同时，这也解释了你看到的_isBulkCopyingInProgress为true的情况：前一批的异步插入还没结束，下一批又尝试调用WriteToServerAsync，但此时SqlBulkCopy还在处理上一批数据，所以这个字段还是true。

解决方案 #

要解决这个问题，需要把整个调用链改成异步的，确保每一批插入完成后再处理下一批：

1. 修改ReadXMLInBatches为异步方法，并在调用processBatch时加上await：

public async Task ReadXMLInBatches(Func<DataTable, Task> processBatch) { 
    int batchSize = 10000; 
    var table = new DataTable(); 
    foreach (var col in columnNames) { 
        table.Columns.Add(col); 
    } 
    using (var reader = new StreamReader(pathToXml, Encoding.GetEncoding(encoding))) 
    using (var xmlReader = XmlReader.Create(reader)) { 
        string lastElement = null; 
        DataRow lastRow = null; 
        while (xmlReader.Read()) { 
            switch (xmlReader.NodeType) { 
                case XmlNodeType.Element: 
                    if (xmlReader.Name == "RECORD") { 
                        if (table.Rows.Count >= batchSize) { 
                            // 等待当前批次插入完成再清空表
                            await processBatch(table); 
                            table.Rows.Clear(); 
                        } 
                        lastRow = table.Rows.Add(); 
                    } 
                    lastElement = xmlReader.Name; 
                    break; 
                case XmlNodeType.Text: 
                    ReadMember(lastRow, lastElement, xmlReader.Value); 
                    break; 
            } 
        } 
        if (table.Rows.Count > 0) { 
            await processBatch(table); 
            table.Rows.Clear(); 
        } 
    } 
}

2. 修改UpdateDBWithXML的参数和调用逻辑，确保等待readXmlInBatches完成：

public async Task UpdateDBWithXML(Func<Func<DataTable, Task>, Task> readXmlInBatches, string hashKey, string hash) { 
    using (var transaction = this.Context.Database.BeginTransaction(IsolationLevel.ReadUncommitted)) 
    using (var bulk = new SqlBulkCopy((SqlConnection)this.Connection, SqlBulkCopyOptions.Default, (SqlTransaction)transaction.UnderlyingTransaction)) { 
        bulk.DestinationTableName = "LegalContractorTemps"; 
        // 等待整个批量读取和插入流程完成
        await readXmlInBatches(async (DataTable table) => { 
            if (bulk.ColumnMappings.Count == 0) { 
                foreach (DataColumn column in table.Columns) { 
                    bulk.ColumnMappings.Add(new SqlBulkCopyColumnMapping(column.ColumnName, column.ColumnName)); 
                } 
            } 
            await bulk.WriteToServerAsync(table); 
        }); 
        await this.Context.Database.ExecuteSqlCommandAsync( 
            "EXECUTE dbo.LegalContractorsDataSynchronize @hashKey, @hash", 
            new SqlParameter("@hashKey", hashKey), 
            new SqlParameter("@hash", hash) 
        ); 
        transaction.Commit(); 
    } 
}

为什么这样能解决问题？ #

修改后，每一批数据的插入操作都会等待完成之后，才会执行table.Rows.Clear()和下一批数据的写入。这样SqlBulkCopy在处理数据时，DataTable的行始终是有效的，不会出现行被提前移除的情况，同时也能保证_isBulkCopyingInProgress字段在每一批插入完成后变回false，不会出现并发调用的问题。

内容的提问来源于stack exchange，提问作者hellouworld