Tengo un desafío aquí, y me está dando dolor de cabeza encontrar una solución.
Tengo una List de algo, y ejecuto Parallel.ForEach basado en eso:
List<Customer> customers = GetNotProcessedCostumer(); Parallel.ForEach(customers, new ParallelOptions {MaxDegreeOfParallelism = 2}, cust=> { ExecuteSomething(cust); }); El problema aquí es que necesito volver a llamar a GetNotProcessedCostumer para verificar si hay nuevos elementos no procesados disponibles en la base de datos, mientras este paralelo aún se está ejecutando. Volver a llamar al método está bien, pero, ¿cómo puedo insertar nuevos elementos en la List que el paralelo ya está usando?
En otras palabras, List<Customer> está vivo, y necesito insertar elementos todo el tiempo en él e intentar usar los subprocesos disponibles en el Parallel existente. Echar un vistazo:
List<Customer> customers = GetNotProcessCustomer // get not processed customers from database Parallel.ForEach(customers) // ...... Start the parallel ... customer.Add(GetNotProcessCustomer()) // Read database again.."Oye Parallel, ¿tienes algún hilo disponible?" Si es así, úsalo.
Puedo aceptar otros enfoques e ideas, como Threads , ThreadPool .........
¿Podría alguien ayudarme por favor?
Probablemente haya mejores formas de hacer este trabajo que la clase Parallel , con un ActionBlock<Customer> de la biblioteca TPL Dataflow como el candidato más prometedor. Pero si desea hacer su trabajo utilizando el conocimiento que ya tiene, puede alimentar el ciclo paralelo con una secuencia IEnumerable<Customer> diferida en lugar de una List<Customer> materializada. Esta secuencia consultará la base de datos y generará los clientes no procesados en un ciclo interminable. Podría ser una buena idea agregar un Task.Delay en la mezcla, para garantizar que la base de datos no se consulte con más frecuencia que cada X segundos.
IEnumerable<Customer> GetNotProcessedCustomersNonStop( CancellationToken cancellationToken = default) { while (true) { var delayTask = Task.Delay(TimeSpan.FromSeconds(5), cancellationToken); foreach (var customer in GetNotProcessedCustomers()) yield return customer; delayTask.GetAwaiter().GetResult(); } } Agregar un CancellationToken en la mezcla probablemente también sea una buena idea, porque eventualmente querrás detener el ciclo, ¿no es así?
Si no está familiarizado con las secuencias enumerables diferidas y la declaración de yield , puede consultar este documento: Iteradores
Un último detalle importante es decirle a la clase Parallel que no desea que haga cosas extravagantes, como enumerar con avidez el enumerable y almacenar en caché sus elementos. Desea que capte al próximo cliente solo cuando esté listo para procesarlo. Puedes hacerlo lanzando un Partitioner.Create en la mezcla. Poniendo todo junto:
var cts = new CancellationTokenSource(); var source = Partitioner.Create(GetNotProcessedCustomersNonStop(cts.Token), EnumerablePartitionerOptions.NoBuffering); var parallelOptions = new ParallelOptions() { MaxDegreeOfParallelism = 2, CancellationToken = cts.Token, }; Parallel.ForEach(source, parallelOptions, customer => { ProcessCustomer(customer); }); //cts.Cancel(); // eventually...