Comment exécuter des lots parallèles d'une liste de saut d'une liste?

Essayez une méthode d'aide comme celle-ci:

public List<Order> BatchOrders(List<Order> orders)
{
    var completedOrders = new ConcurrentBag<List<Order>>();

    var opts = new ParallelOptions() { MaxDegreeOfParallelism = 4 };

    int batchSize = 2000;
    var batches = orders.Batch(batchSize).ToList();
    Parallel.For(0,batches.Count,opts, 
        batchNum =>
           {
               var batch = batches[batchNum];
               var startId = batchNum * batchSize;
               completedOrders.Add(CreateOrders(batch, startId));
           });

    return completedOrders.SelectMany(c => c).ToList();
}

Ensuite, pour gérer chaque lot en parallèle, quelque chose comme:

public List<Order> BatchOrders(List<Order> orders)
{
    var completedOrderBatches = new ConcurrentBag<List<Order>>();

    var opts = new ParallelOptions() { MaxDegreeOfParallelism = 4 };

    Parallel.ForEach(orders.Batch(),opts,  
        batch =>
           {
               completedOrderBatches.Add(CreateOrders(batch));
           });

    return completedOrderBatches.SelectMany(c => c).ToList();
}

Si vous avez besoin le numéro de lot dans CreateOrders, vous pouvez alors utiliser Parallel.For au lieu de Paralle.ForEach. Tout comme utiliser pour au lieu de foreach.

public static class BatchExtensions
{
    public static IEnumerable<List<T>> Batch<T>(this IEnumerable<T> col, int batchSize = 2000)
    {
        var batch = new List<T>(batchSize);
        foreach (var o in col)
        {
            batch.Add(o);
            if (batch.Count == batchSize)
            {
                var rc = batch;
                batch = new List<T>(batchSize);
                yield return rc;
            }
        }
        if (batch.Count > 0)
        {
            yield return batch;
        }
    }
}

Quelque chose comme ça?

public List<Order> BatchOrders(List<Order> orders)
{        
    int batchSize = 2000;

    List <Task<IEnumerable<Order>>> tasks = new List<Task<IEnumerable<Order>>>();

    //Create orders in batch
    for (int i = 0; i < orders.Count; i += batchSize)
    {
        var batchOrders = orders.Skip(i).Take(batchSize).ToList();

        // Run CreateOrders as a task and store the task
        tasks.Add(Task.Run(() => CreateOrders(batchOrders)));
    }

    var allTasks = tasks.ToArray();

    // Wait till all the tasks are complete
    Task.WaitAll(allTasks);

    var completedOrders = new List<Order>();

    //Merge the results
    foreach (var task in allTasks)
        completedOrders.AddRange(task.Result);

    return completedOrders;
}

Je pense que les réponses qui sont déjà fournies sont excellentes!

Je voudrais partager les miennes, car j'adore le code déclaratif et concis.

Veuillez noter que bien que ce code soit meilleur lisibilité , elle a une efficacité d'exécution moins bonne , car la collection source est énumérée autant de fois qu'il y a de lots.

Les méthodes d'extension sont définies sur IReadonlyCollection et pas sur IEnumerable pour exclure les énumérations évaluées paresseusement, qui chargent les données à partir de sources de données externes.

Premièrement, quelques assistants d'extension

    async Task<IEnumerable<CreatedOrder>> BatchOrders(List<Order> orders, int batchSize)
    {
        var batches = 
            orders
            .Batch(batchSize)
            .Select(batch => Task.Run(() => CreateOrders(batch)))
            .ToArray();

        var result = (await Task.WhenAll(batches)).SelectMany(x=>x);

        return result;
    }

Maintenant, votre méthode est plutôt courte et concise

public static class BatchExtensions
{
    public static IEnumerable<T> TakePart<T>(this IReadOnlyCollection<T> data, int batchNumber, int batchSize) =>
        data
        .Skip(batchNumber * batchSize)
        .Take(batchSize);

    public static IEnumerable<IEnumerable<T>> Batch<T>(this IReadOnlyCollection<T> data, int batchSize) =>
        Enumerable
        .Range(0, data.Count / batchSize)
        .Select(index => TakePart(data, index, batchSize));
}

Voici comment vous pouvez le faire avec l'aide de la Bibliothèque TPL Dataflow . Vous auriez besoin de deux blocs, un BatchBlock pour effectuer le traitement par lots et un TransformManyBlock pour traiter chaque lot.

List<Order> BatchOrders(List<Order> orders,
    CancellationToken cancellationToken = default)
{
    var batchBlock = new BatchBlock<Order>(batchSize: 2000);

    var tranformBlock = new TransformManyBlock<Order[], Order>(batch =>
    {
        return CreateOrders(batch.ToList());
    }, new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = 5,
        CancellationToken = cancellationToken
    });

    batchBlock.LinkTo(tranformBlock,
        new DataflowLinkOptions() { PropagateCompletion = true });

    foreach (var order in orders)
    {
        batchBlock.Post(order);
    }
    batchBlock.Complete();

    var results = new List<Order>();
    while (tranformBlock.OutputAvailableAsync().Result)
    {
        while (tranformBlock.TryReceive(out var order))
        {
            results.Add(order);
        }
    }
    tranformBlock.Completion.GetAwaiter().GetResult();
    return results;
}

Cette bibliothèque est intégrée à .NET Core. Les avantages sont:

1. Prise en charge intégrée de la configuration du degré maximal de parallélisme
2. Prise en charge intégrée de l'annulation
3. Échec rapide en cas d'exception (contrairement à Task.WaitAll par exemple)
4. Peut gérer des scénarios plus complexes
5. Peut être converti relativement facilement en asynchrone

Et les inconvénients:

1. Moins familier que les alternatives
2. Récupérer les résultats du bloc final est quelque peu fastidieux
3. Non intégré à .NET Framework (nécessite un package installation)