异步填充 C# 字典

Question

我使用下面的代码循环遍历合同列表并获取每个合同的费率列表。然后它将结果存储在字典中。

/// <summary>
/// Get the Rates for the Job keyed to the id of the Organisation
/// </summary>
async Task<Dictionary<int, IReadOnlyList<Rate>>> GetRatesAsync(int jobId)
{
    var jobContracts = await _contractClient.GetJobContractsByJobAsync(jobId);
    var result = new Dictionary<int, IReadOnlyList<Rate>>();
    foreach (var jobContract in jobContracts)
    {
        result.Add(jobContract.Contract.Organisation.Id, await _contractClient.GetContractRatesByContractAsync(jobContract.Contract.Id));
    }
    return result;
}

Resharper 很有帮助地建议我“循环可以转换为 LINQ 表达式”。但是，Resharper 自动生成的修复程序（如下）无法编译。

return jobContracts.ToDictionary(jobContract => jobContract.Contract.Organisation.Id, jobContract => await _contractClient.GetContractRatesByContractAsync(jobContract.Contract.Id));

进而

错误是The await operator can only be used within an async lamba expression。因此，插入一个异步并将其更改为以下内容：

return jobContracts.ToDictionary(jobContract => jobContract.Contract.Organisation.Id, async jobContract => await _contractClient.GetContractRatesByContractAsync(jobContract.Contract.Id));

进而

现在出现错误Cannot implicitly convert type 'System.Collections.Generic.Dictionary<int, System.Threading.Tasks.Task<System.Collections.Generic.IReadOnlyList<Rate>>>' to 'System.Collections.Generic.Dictionary<int, System.Collections.Generic.IReadOnlyList<Rate>>'，它希望我将签名更改为：

async Task<Dictionary<int, Task<IReadOnlyList<Rate>>>> GetContractRatesAsync()

进而

Dictionary<int, IReadOnlyList<Rate>>最后的更改现在可以编译，但我所有的调用方法都期望等待一个 Task ，而不是一个Dictionary<int, Task<IReadOnlyList<Rate>>.

有什么方法可以等待将其转换为 a 吗Dictionary<int, IReadOnlyList<Rate>>？或者其他异步获取数据的方法？

Answer 1

实际上，通用方法可能相当有趣，所以我将发布一个解决方案。您的初始实现存在一个性能问题，如果您愿意，您可以解决它，并且您的应用程序域允许这样做。

如果您想异步地将项目添加到字典中，那么可能需要允许这些异步项目并行运行。您希望从接受可枚举的键和值的方法开始。这些值应该是生成任务的函数，而不是任务。这样您就不会同时运行所有任务。

然后，您可以使用信号量来控制同时启动的任务数量。当您希望所有任务同时开始时，下面还有一个后备功能。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace _43909210
{
    public static class EnumerableExtensions
    {
        private static async Task<Dictionary<TKey, TValue>> ToDictionary<TKey,TValue>
            (this IEnumerable<(TKey key, Func<Task<TValue>> valueTask)> source)
        {

            var results = await Task.WhenAll
                ( source.Select( async e => (  key: e.key, value:await e.valueTask() ) ) );

            return results.ToDictionary(v=>v.key, v=>v.value);
        }

        public class ActionDisposable : IDisposable
        {
            private readonly Action _Action;
            public ActionDisposable(Action action) => _Action = action;
            public void Dispose() => _Action();
        }

        public static async Task<IDisposable> Enter(this SemaphoreSlim s)
        {
            await s.WaitAsync();
            return new ActionDisposable( () => s.Release() );
        }


        /// <summary>
        /// Generate a dictionary asynchronously with up to 'n' tasks running in parallel.
        /// If n = 0 then there is no limit set and all tasks are started together.
        /// </summary>
        /// <returns></returns>
        public static async Task<Dictionary<TKey, TValue>> ToDictionaryParallel<TKey,TValue>
            ( this IEnumerable<(TKey key, Func<Task<TValue>> valueTaskFactory)> source
            , int n = 0
            )
        {
            // Delegate to the non limiting case where 
            if (n <= 0)
                return await ToDictionary( source );

            // Set up the parallel limiting semaphore
            using (var pLimit = new SemaphoreSlim( n ))
            {
                var dict = new Dictionary<TKey, TValue>();

                // Local function to start the task and
                // block (asynchronously ) when too many
                // tasks are running
                async Task
                    Run((TKey key, Func<Task<TValue>> valueTask) o)
                {
                    // async block if the parallel limit is reached
                    using (await pLimit.Enter())
                    {
                        dict.Add(o.key, await o.valueTask());
                    }
                }

                // Proceed to start the tasks
                foreach (var task in source.Select( Run ))
                    await task;

                // Wait for all tasks to finish
                await pLimit.WaitAsync();
                return dict;
            }

        }
    }

}