c# - 如何处理匿名任务的异常？

Question

我有一种方法可以从服务器中分块提取数据并将其返回进行处理。我做了一些测量，发现在后台下载 block 并通过BlockingCollection<T>返回它们要快得多。 。这允许客户端和服务器同时工作，而不是互相等待。

public static IEnumerable<DataRecord> GetData(String serverAddress, Int64 dataID)
{
    BlockingCollection<DataRecord> records = new BlockingCollection<DataRecord>();

    Task.Run(
        () =>
        {
            Boolean isMoreData = false;
            do
            {
                // make server request and process response
                // this block can throw

                records.Add(response.record);
                isMoreData = response.IsMoreData;
            }
            while (isMoreData);

            records.CompleteAdding();
        });

    return records.GetConsumingEnumerable();
}

调用者(C++/CLI 库)应该知道发生了异常，以便它可以重试或酌情退出。在最小程度地更改返回类型的同时将异常传播给调用者的最佳方法是什么？

Answer 1

这就是为什么“即发即忘”任务通常不是一个好主意。在您的情况下，这是一个更糟糕的想法，因为您没有将添加内容包装在 try/catch 中，并在 finally 中使用 records.CompleteAdding code> 阻塞意味着从 GetConsumingEnumerable 对枚举器的 MoveNext 调用最终将无限期阻塞 - 这很糟糕。

如果您完全在 C# 范围内操作，解决方案很简单:更好地分离关注点。您去掉 BlockingCollection 位并在其所属的位置运行它:在消费者(客户端)中，或在中间管道处理阶段(这最终是您想要实现的目标)中，该阶段将在它始终了解生产者抛出的任何异常。然后，您的 GetData 签名保持不变，但它变成了具有完整异常传播的简单阻塞可枚举:

public static IEnumerable<DataRecord> GetData(String serverAddress, Int64 dataID)
{
    Boolean isMoreData = false;
    do
    {
        // make server request and process response
        // this block can throw

        yield return response.record;
        isMoreData = response.IsMoreData;
    }
    while (isMoreData);
}

然后管道看起来像这样:

var records = new BlockingCollection<DataRecord>();

var producer = Task.Run(() =>
{
    try
    {
        foreach (var record in GetData("http://foo.com/Service", 22))
        {
            // Hand over the record to the
            // consumer and continue enumerating.
            records.Add(record);
        }
    }
    finally
    {
        // This needs to be called even in
        // exceptional scenarios so that the
        // consumer task does not block
        // indefinitely on the call to MoveNext.
        records.CompleteAdding();
    }
});

var consumer = Task.Run(() =>
{
    foreach (var record in records.GetConsumingEnumerable())
    {
        // Do something with the record yielded by GetData.
        // This runs in parallel with the producer,
        // So you get concurrent download and processing
        // with a safe handover via the BlockingCollection.
    }
});

await Task.WhenAll(producer, consumer);

现在您可以鱼与熊掌兼得:当 GetData 生成记录时，处理是并行进行的，并且 await 生产者任务会传播任何异常，而在 finally 中调用 CompleteAdding 可确保您的消费者不会无限期地陷入阻塞状态。

由于您正在使用 C++，上述内容在一定程度上仍然适用(也就是说，正确的做法是在 C++ 中重新实现管道)，但实现可能不是那么漂亮，而且您的方式已经同意了，您的答案很可能是首选解决方案，即使由于未观察到的任务而确实感觉像是黑客。我真的无法想象它实际上会出错的场景，因为由于新引入的try/catch，CompleteAdding总是被调用。

显然，另一个解决方案是将处理代码移至您的 C# 项目，这可能会也可能不会，具体取决于您的体系结构。