Recientemente me hicieron una pregunta en una entrevista y realmente me hizo pensar.
Estoy tratando de comprender y aprender más sobre subprocesos múltiples, paralelismo y concurrencia, y rendimiento.
El escenario es que tiene una lista de rutas de archivo. Los archivos se guardan en su disco duro o en el almacenamiento de blobs. Ha leído los archivos y los ha almacenado en una base de datos. ¿Cómo lo haría de la manera más óptima?
Las siguientes son algunas de las formas en las que podría pensar:
La forma más sencilla es recorrer la lista y realizar esta tarea secuencialmente.
Foreach(var filePath in filePaths) { ProcessFile(filePath); } public void ProcessFile(string filePath) { var file = readFile(filePath); storeInDb(file); }La segunda forma en la que podría pensar es en crear varios hilos, tal vez:
Foreach(var filePath in filePaths) { Thread t = new Thread(ProcessFIle(filePath)); t.Start(); } (not sure if the above code is correct.)La tercera forma es usar async await
List<Tasks> listOfTasks; Foreach(var filePath in filePaths) { var task = ProcessFile(filePath); listOfTasks.Add(task); } Task.WhenAll(listOftasks); public async void ProcessFile(string filePath) { var file = readFile(filePath); storeInDb(file); }La cuarta vía es paralela. Para:
Parallel.For(0,filePaths.Count , new ParallelOptions { MaxDegreeOfParallelism = 10 }, i => { ProcessFile(filePaths[i]); });Cuáles son las diferencias entre ellos. ¿Cuál sería más adecuado para el trabajo y hay algo mejor?
También puede usar Reactive Framework de Microsoft (también conocido como Rx) - NuGet System.Reactive y agregar using System.Reactive.Linq; - entonces puedes hacer esto:
IObservable<string> query = from filePath in filePaths.ToObservable() from file in Observable.Start(() => ReadFile(filePath)) from db in Observable.Start(() => StoreInDb(file)) select filePath; IDisposable subscription = query .Subscribe( filePath => Console.WriteLine($"{filePath} Processed."), () => Console.WriteLine("Done."));Escribí un método de extensión simple para ayudar a iniciar tareas asíncronas, limitar la cantidad de simultaneidad y esperar a que se completen;
public static async Task WhenAll(this IEnumerable<Task> tasks, int batchSize) { var started = new List<Task>(); foreach(var t in tasks) { started.Add(t); if (started.Count >= batchSize) { var ended = await Task.WhenAny(started); started.Remove(ended); } } await Task.WhenAll(started); }Entonces querrá un método para transmitir el contenido del archivo directamente a la base de datos. Por ejemplo;
async Task Process(string filename){ using var stream = File.OpenRead(filename) // TODO connect to the database var sqlCommand = ...; sqlCommand.CommandText = "update [table] set [column] = @stream"; sqlCommand.Parameters.Add(new SqlParameter("@stream", SqlDbType.VarBinary) { Value = stream }); await sqlCommand.ExecuteNonQueryAsync(); } IEnumerable<string> files = ...; await files.Select(f => Process(f)).WhenAll(20);¿Es este el mejor enfoque? Probablemente no. Dado que es demasiado fácil hacer un mal uso de esta extensión. Iniciar tareas accidentalmente varias veces, o iniciarlas todas a la vez.