C# AsyncEnumerable 运行/等待多个任务永远不会完成

Question

我想要一个接收 Task<bool> 的函数并在 X 任务中运行它。

为此，我编写了以下代码:

public static class RetryComponent
{
    public static async Task RunTasks(Func<Task<bool>> action, int tasks, int retries, string method)
    {
        // Running everything
        var tasksPool = Enumerable.Range(0, tasks).Select(i => DoWithRetries(action, retries, method)).ToArray();
        await Task.WhenAll(tasksPool);
    }

    private static async Task<bool> DoWithRetries(Func<Task<bool>> action, int retryCount, string method)
    {
        while (true)
        {
            if (retryCount <= 0)
                return false;

            try
            {
                bool res = await action();
                if (res)
                    return true;
            }
            catch (Exception e)
            {
                // Log it
            }

            retryCount--;
            await Task.Delay(200); // retry in 200
        }
    }
}

以及下面的执行代码:

BlockingCollection<int> ints = new BlockingCollection<int>();
foreach (int i in Enumerable.Range(0, 100000))
{
    ints.Add(i);
}
ints.CompleteAdding();

int taskId = 0;
var enumerable = new AsyncEnumerable<int>(async yield =>
{
    await RetryComponent.RunTasks(async () =>
    {
        try
        {
            int myTaskId = Interlocked.Increment(ref taskId);

            // usually there are async/await operations inside the while loop, this is just an example

            while (!ints.IsCompleted)
            {
                int number = ints.Take();

                Console.WriteLine($"Task {myTaskId}: {number}");
                await yield.ReturnAsync(number);
            }
        }
        catch (InvalidOperationException)
        {
            return true;
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
            throw;
        }

        return true;
    }, 10, 1, MethodBase.GetCurrentMethod().Name);
});

await enumerable.ForEachAsync(number =>
{
    Console.WriteLine(number);
});

哪里AsyncEnumerable来自System.Collections.Async .

控制台显示任务 10:X(其中 x 是列表中的数字..)。

当我删除 AsyncEnumerable一切都按预期工作(所有任务都在打印并且执行结束)..出于某种原因，我很长时间都找不到，使用 AsyncEnumerable只会破坏一切(在我的主要代码中，我需要它来使用 AsyncEnumerable .. 可伸缩性的东西..)意味着代码永远不会停止，只有最后一个任务 (10) 正在打印。当我添加更多日志时，我看到任务 1-9 永远不会完成。

因此，为了清楚起见，我希望让多个任务执行异步操作并将结果生成给充当管道的单个 AsyncEnumerable 对象。 (这就是想法..)

Answer 1

问题在于枚举器/生成器模式是顺序的，但您正在尝试执行多生产者、单一消费者模式。由于您使用嵌套的匿名函数，并且堆栈溢出不显示行号，因此很难准确描述我指的是代码的哪一部分，但无论如何我都会尝试。

AsyncEnumerable 的工作方式基本上是等待生产者产生一个值，然后等待消费者使用该值，然后重复。它不支持生产者和消费者以不同的速度运行，因此我说这种模式是顺序的。它没有生产项目队列，only the current value . ReturnAsync does not wait为了让消费者使用该值，您应该等待它返回的任务，这会向您发出准备就绪的信号。因此我们可以得出结论，它不是线程安全的。

但是，RetryComponent.RunTasks 并行运行 10 个任务，并且该代码调用 yield.ReturnAsync 而不检查是否有其他人已经调用它，如果是，则该任务是否已调用完全的。由于 Yield 类仅存储当前值，您的 10 个并发任务会覆盖当前值，而无需等待 Yield 对象准备好接收新值，因此其中 9 个任务丢失并且永远不会等待.由于从未等待这 9 个任务，因此这些方法永远不会完成并且 Task.WhenAll 永远不会返回，整个调用堆栈中的任何其他方法也不会返回。

I created an issue on github建议他们改进他们的库以在发生这种情况时抛出异常。如果他们实现它，您的 catch block 会将消息写入控制台并重新抛出错误，将任务置于故障状态，这将允许 task.WhenAll 完成，因此您的程序不会挂了。

您可以使用多线程同步 API 来确保一次只有一个任务调用 yield.ReturnAsync 并等待返回任务。或者您可以避免使用多生产者模式，因为单个生产者很容易成为枚举器。否则，您将需要完全重新考虑如何实现多生产者模式。我建议 TPL Dataflow它内置于 .NET Core，并在 .NET Framework 中作为 NuGet 包提供。