¿Hay alguna manera de usar la Biblioteca de tareas paralelas (TPL) con SQLDataReader?

Question

Me gusta la simplicidad de los métodos de extensión Parallel.For y Parallel.ForEach en el TPL. Me preguntaba si había una forma de aprovechar algo similar o incluso con Tareas ligeramente más avanzadas.

A continuación se muestra un uso típico de SqlDataReader, y me preguntaba si era posible y, en caso afirmativo, cómo reemplazar el ciclo while a continuación con algo en el TPL. Debido a que el lector no puede proporcionar un número fijo de iteraciones, el método de extensión For no es posible, lo que deja el tratamiento de las tareas que reuniría. Esperaba que alguien ya haya abordado esto y haya averiguado algunas cosas que hacer y no con ADO.net.

using (SqlConnection conn = new SqlConnection("myConnString")) 
using (SqlCommand comm = new SqlCommand("myQuery", conn)) 
{ 
    conn.Open(); 

    SqlDataReader reader = comm.ExecuteReader(); 

    if (reader.HasRows) 
    { 
     while (reader.Read()) 
     { 
      // Do something with Reader 
     } 
    } 
} 

Answer 1

Ya casi está allí. Envuelve el código que envió en una función con esta firma:

IEnumerable<IDataRecord> MyQuery() 

y luego vuelva a colocar su código // Do something with Reader con esto:

yield return reader; 

Ahora usted tiene algo que funciona en un solo hilo. Desafortunadamente, a medida que lee los resultados de la consulta, devuelve una referencia al mismo objeto cada vez, y el objeto simplemente se muta para cada iteración. Esto significa que si intenta ejecutarlo en paralelo, obtendrá resultados realmente extraños, ya que las lecturas paralelas modifican el objeto utilizado en diferentes subprocesos. Necesita código para tomar una copia del registro para enviar a su bucle paralelo.

En este punto, sin embargo, lo que me gusta hacer es omitir la copia extra del registro e ir directamente a una clase fuertemente tipada. Más que eso, me gusta usar un método genérico para hacerlo:

IEnumerable<T> GetData<T>(Func<IDataRecord, T> factory, string sql, Action<SqlParameterCollection> addParameters) 
{ 
    using (var cn = new SqlConnection("My connection string")) 
    using (var cmd = new SqlCommand(sql, cn)) 
    { 
     addParameters(cmd.Parameters); 

     cn.Open(); 
     using (var rdr = cmd.ExecuteReader()) 
     { 
      while (rdr.Read()) 
      { 
       yield return factory(rdr); 
      } 
     } 
    } 
} 

Suponiendo sus métodos de fábrica crean una copia como era de esperar, este código debe ser seguro para su uso en un bucle Parallel.ForEach. Al llamar al método sería algo como esto (suponiendo una clase a un empleado con un método de fábrica estática denominada "Crear"):

var UnderPaid = GetData<Employee>(Employee.Create, 
     "SELECT * FROM Employee WHERE AnnualSalary <= @MinSalary", 
     p => { 
      p.Add("@MinSalary", SqlDbType.Int).Value = 50000; 
     }); 
Parallel.ForEach(UnderPaid, e => e.GiveRaise()); 

Actualización importante:
no estoy tan seguro en este código como yo una vez fueUn hilo separado podría mutar el lector mientras otro hilo está en el proceso de hacer su copia. Podría ponerle un candado, pero también me preocupa que otro hilo pueda llamar al lector una vez que el original haya llamado a Read() pero antes de que empiece a hacer la copia. Por lo tanto, la sección crítica aquí consiste en todo el ciclo while ... y en este punto, vuelves a un único subproceso nuevamente. Espero que haya una manera de modificar este código para que funcione como se espera para los escenarios de subprocesos múltiples, pero se necesitará más estudio.

Answer 2

Vas a tener dificultades para reemplazar ese bucle while directamente. SqlDataReader no es una clase segura para subprocesos, por lo que no puede usarla directamente desde varios subprocesos.

Dicho esto, podría procesar los datos que lee utilizando el TPL. Hay algunas opciones, aquí. Lo más fácil podría ser hacer su propia implementación IEnumerable<T> que funcione en el lector, y devuelva una clase o estructura que contenga sus datos. A continuación, puede utilizar PLINQ o una declaración Parallel.ForEach para procesar sus datos en paralelo:

public IEnumerable<MyDataClass> ReadData() 
{ 
    using (SqlConnection conn = new SqlConnection("myConnString")) 
    using (SqlCommand comm = new SqlCommand("myQuery", conn)) 
    { 
     conn.Open(); 

     SqlDataReader reader = comm.ExecuteReader(); 

     if (reader.HasRows) 
     { 
      while (reader.Read()) 
      { 
       yield return new MyDataClass(... data from reader ...); 
      } 
     } 
    } 
} 

vez que tenga ese método, se puede procesar de manera directa, a través de PLINQ o TPL:

Parallel.ForEach(this.ReadData(), data => 
{ 
    // Use the data here... 
}); 

O:

this.ReadData().AsParallel().ForAll(data => 
{ 
    // Use the data here... 
});