XDocument 到 DataTable 异步

Posted 2023-04-12

【中文标题】XDocument 到 DataTable 异步

【英文标题】：XDocument to DataTable async

【发布时间】：2021-11-04 17:30:26

【问题描述】：

我正在将 100-500 个 XML 文件中的数据加载到 DataTable 以填充 Infragistics UltraGrid。这些文件小至 100K，大至 2MB。我一直在寻找加快加载时间并考虑异步的方法，但数据表不是线程安全的。

看流程，加载xdoc大约需要2/3的时间，另外1/3是在读取xdoc和向表中添加数据。

有没有办法使用异步并以某种方式加载下一个 XDocument 而前一个被读取并加载到数据表中？我看过，但我没有看到一个很好的方法来为每个人做到这一点。还有其他一些我应该考虑的策略吗？

这是我正在做的简化版本：

private void openXMLs()
    
    OpenFileDialog openFileDialog1 = new OpenFileDialog
    
        Title = "Browse XML files",
        CheckFileExists = true,
        CheckPathExists = true,
        DefaultExt = "xml",
        Filter = @"XML files (*.xml)|*.xml",
        FilterIndex = 2,
        RestoreDirectory = true,
        ReadOnlyChecked = true,
        ShowReadOnly = true,
        Multiselect = true
    ;

    if (openFileDialog1.ShowDialog() == DialogResult.OK)
    
        foreach (String file in openFileDialog1.FileNames)
        
            XDocument xdoc = XDocument.Load(file)
            loadData(xdoc)
        
    

        
private void loadData(XDocument xdoc)
        
    var query = xdoc.Descendants("root").AsEnumerable()
    .Select(c => new
    
        id = c.Element("ID").Value,
        firstname = c.Element("firstname").Value,
        lastname = c.Element("lastname").Value,
        state = c.Element("state").Value,

    );
    
    foreach (var item in query)
    
        _dt.Rows.Add(
            item.id,
            item.firstname,
            item.lastname,
            item.state
        );
    

【问题讨论】：

有些相关：How to enumerate an IAsyncEnumerable<T> and invoke an async action for each element, allowing concurrency for each iteration/action pair?

我正在将 100-500 个 XML 文件中的数据加载到 DataTable 中

很遗憾，这是我必须工作的环境。

【参考方案1】：

当然，这个问题没有单一的答案。您可以通过多种方式做到这一点，这里是一种：

private async Task openXMLs() 
    
    // .. dialog ..

    var loadTasks = new List<Task>();        
    if (openFileDialog1.ShowDialog() == DialogResult.OK)
    
        foreach (String file in openFileDialog1.FileNames)
        
            var xmlFile = XmlReader.Create(file);
            XDocument xdoc = await XDocument.LoadAsync(xmlFile, LoadOptions.None, CancellationToken.None);
            loadTasks.Add(Task.Run(() => loadData(xdoc));
        
    

    await Task.WhenAll(loadTasks);

但是，数据表不是线程安全的，您需要锁定对它的访问。这也不能保证添加文档的顺序。

private void loadData(XDocument xdoc)
    
    var query = xdoc.Descendants("root").AsEnumerable()
    .Select(c => new
    
        id = c.Element("ID").Value,
        firstname = c.Element("firstname").Value,
        lastname = c.Element("lastname").Value,
        state = c.Element("state").Value,

    );
    
    lock(_dt) 
        foreach (var item in query)
        
            _dt.Rows.Add(
                item.id,
                item.firstname,
                item.lastname,
                item.state
             );
        
    

这将在下一个从磁盘读取时加载当前文档。

【讨论】：

谢谢你，我无法完全理解如何做到这一点，但现在看到这个很有意义。我做了一个初始测试，看起来这将使我的加载时间快两倍，所以这正是我所需要的。请注意，我不能等待 XDocument.LoadAsync。我错过了什么吗？将其保留为 XDocument.Load 会有问题吗？

不幸的是，LoadAsync 似乎没有过载来接受文件参数，但我认为你可以使用XmlReader。更新了我的答案。

【参考方案2】：

您可以使用如下所示的ParallelForEachPairwise 扩展方法并行加载文件系统中的下一个XDocument，同时将上一个文档加载到DataTable：

/// <summary>
/// Invokes two consecutive actions on each element in the source sequence.
/// The first action is invoked in parallel with the second action for the
/// previous element, and with fetching the sequence's next element.
/// </summary>
public static void ParallelForEachPairwise<TSource, TResult>(
    this IEnumerable<TSource> source,
    Func<TSource, TResult> offloadedAction,
    Action<TResult> synchronousAction)

    using var enumerator = source.GetEnumerator();
    if (!enumerator.MoveNext()) return;
    Task<TResult> task = Offload(enumerator.Current);
    try
    
        while (enumerator.MoveNext())
        
            TResult result = task.GetAwaiter().GetResult();
            task = Offload(enumerator.Current);
            synchronousAction(result);
        
        synchronousAction(task.GetAwaiter().GetResult());
    
    finally  task.GetAwaiter().GetResult();  // Prevent fire-and-forget

    Task<TResult> Offload(TSource item) => Task.Run(() => offloadedAction(item));

ParallelForEachPairwise 接受两个委托作为参数。第一个委托 (offloadedAction) 通过Task.Run 方法卸载到ThreadPool。在当前线程上同步调用第二个委托 (synchronousAction)。这些委托彼此并行调用，但每个单独委托的重复调用是顺序的。因此，您不必担心线程安全，前提是每个委托不会产生其他委托可见的副作用。

使用示例：

openFileDialog1.FileNames.ParallelForEachPairwise(file =>

    return XDocument.Load(file);
, xdoc =>

    loadData(xdoc);
);

...或更简洁地说：

openFileDialog1.FileNames.ParallelForEachPairwise(XDocument.Load, loadData);

【讨论】：

这看起来是一个有趣的方法，我也会尝试测试它，看看它如何与其他方法叠加。谢谢！

@Fisher8370 这是一种保守的并行方法。您可以期望它最多可将性能提升 2 倍，但它几乎是免费的。与简单的顺序方法相比，存储驱动程序和 CPU 不会因大量额外工作而繁重。