C＃EF Core-分区输入和并行执行查询

问题描述

我有以下代码，我要在其中检索给定雇员ID列表的雇员信息。如果计数超过一百万，我已经进行了验证以引发异常。在大多数情况下，请求将小于200K，因此我将请求分为4个分区，每个分区包含相等数量的员工ID。所有4个分区并行执行，并在Task.WhenAll之后合并在一起。有人可以给我一些进一步改进的提示吗？我查看了https://houseofcat.io/tutorials/csharp/async/parallelforeachasync和https://medium.com/@alex.puiu/parallel-foreach-async-in-c-36756f8ebe62，但无法使其正常工作。下面提到的代码可以工作，但其硬编码可分为4个分区。我如何使其与最大并行度设置为50的动态分区更加并行？如果输入的是10万个ID，我想分成10个分区并并行执行全部10个分区。

 public class Service
{
    private async Task<List<EmployeeEntity>> GetInfo(List<long> input)
    {
        var breakup = input.Split(4);

        var result1Task = GetResult(breakup.First().ToList());
        var result2Task = GetResult(breakup.Skip(1).Take(1).First().ToList());
        var result3Task = GetResult(breakup.Skip(2).Take(1).First().ToList());
        var result4Task = GetResult(breakup.Skip(3).Take(1).First().ToList());

        await Task.WhenAll(result1Task,result2Task,result3Task,result4Task);

        List<EmployeeEntity> result1 = await result1Task;
        List<EmployeeEntity> result2 = await result2Task;
        List<EmployeeEntity> result3 = await result3Task;
        List<EmployeeEntity> result4 = await result4Task;

        return result1.Union(result2.Union(result3.Union(result4))).ToList();
    }

    private async Task<List<EmployeeEntity>> GetResult(List<long> employees)
    {
        using var context = new MyAppDBContext();
        var EmployeeBand = await context.EmployeeBand.Where(x => employees.Contains(x.EmployeeId)).ToListAsync();
        var EmployeeClient = await context.EmployeeClient.Where(x => employees.Contains(x.EmployeeId)).ToListAsync();
        return await context.Employee.Where(x => employees.Contains(x.EmployeeId)).ToListAsync();
    }
}

public static class ExtensionMethods
{
    public static List<List<T>> Split<T>(this List<T> myList,int parts)
    {
        int i = 0;
        var splits = from item in myList
                     group item by i++ % parts into part
                     select part.ToList();
        return splits.ToList();
    }
}

public class EmployeeEntity
{
    public EmployeeEntity()
    {
        EmployeeBands = new HashSet<EmployeeBandEntity>();
        EmployeeClients = new HashSet<EmployeeClientEntity>();
    }

    public long EmployeeId { get; set; }
    public ICollection<EmployeeBandEntity> EmployeeBands { get; set; }
    public ICollection<EmployeeClientEntity> EmployeeClients { get; set; }
}

public class EmployeeBandEntity
{
    public long EmployeeBandId { get; set; }
    public long EmployeeId { get; set; }
    public EmployeeEntity EmployeeEntity { get; set; }
}

public class EmployeeClientEntity
{
    public long EmployeeClientId { get; set; }
    public long EmployeeId { get; set; }
    public EmployeeEntity EmployeeEntity { get; set; }
}

public partial class MyAppDBContext : DbContext
{
    public virtual DbSet<EmployeeEntity> Employee { get; set; }
    public virtual DbSet<EmployeeBandEntity> EmployeeBand { get; set; }
    public virtual DbSet<EmployeeClientEntity> EmployeeClient { get; set; }
}
     

解决方法

我相信您可能对GetResult很有创意，并且可以采用更好的方式进行重写，以使查询类似于where id greater than (and/or less than)而不是ids in (... list)。假设您的GetResult已经以最佳方式实现，而您只需要一种实现最大并行执行的方法，这就是我的解决方案。

private async Task<List<EmployeeEntity>> GetInfo2(List<long> input)
{
    if (input == null)
    {
        return null;
    }

    if (input.Count == 0)
    {
        return new List<EmployeeEntity>();
    }

    var taskList = new List<Task<List<EmployeeEntity>>>();

    foreach (var batch in input.Batch(100))
    {
        taskList.Add(GetResult(batch.ToList()));
    }

    var result = (await Task.WhenAll(taskList)).SelectMany(a => a);

    return result.ToList();
}

这需要以下批处理扩展方法。

public static class Extensions
    {
        public static IEnumerable<IEnumerable<T>> Batch<T>(this IEnumerable<T> source,int size)
        {
            T[] bucket = null;
            var count = 0;

            foreach (var item in source)
            {
                if (bucket == null)
                    bucket = new T[size];

                bucket[count++] = item;

                if (count != size)
                    continue;

                yield return bucket.Select(x => x);

                bucket = null;
                count = 0;
            }

            if (bucket != null && count > 0)
                yield return bucket.Take(count);
        }
    }

您可以找到批量大小的最佳选择。我已经将其硬编码为100，但是您可以根据输入列表的大小或任何其他逻辑来推导它。

,

这是一种管理任务的替代解决方案，而无需编写我为先前的答案（在注释中）建议的信号量限制的东西。我遇到了使用来自ActionBlock名称空间的System.Threading.Tasks.Dataflow的.net开箱即用解决方案。这样可以在不使用信号量苗条等的情况下，将并发保持在所需的最大程度。

Batch扩展名仍然与我之前的答案相同。但是对于第一部分，您将使用如下所示的ActionBlock（而不是Task.WhenAll）。

        private async Task<List<EmployeeEntity>> GetInfo2(List<long> input)
    {
        if (input == null)
        {
            return null;
        }

        if (input.Count == 0)
        {
            return new List<EmployeeEntity>();
        }

        var employeeEntities = new BlockingCollection<EmployeeEntity>();

        var actionBlock = new ActionBlock<List<long>>(async (employeeIds) =>
        {
            var employees = await GetResult(employeeIds);
            employees.ForEach(e => employeeEntities.Add(e));
        },new ExecutionDataflowBlockOptions
        {
            //config this to whatever works best in your situation
            MaxDegreeOfParallelism = 20
        });

        foreach (var batch in input.Batch(100))
        {
            await actionBlock.SendAsync(batch.ToList());
        }

        actionBlock.Complete();
        await actionBlock.Completion;

        return employeeEntities.ToList();
    }