Есть ли способ использовать параллельную библиотеку задачи (TPL) с SQLDatareader?

Question

Мне нравится простота параллельного. Для методов расширения Parallel.Foreach в TPL. Мне было интересно, был ли способ воспользоваться чем-то похожим или даже с немного более продвинутыми задачами.

Ниже приведено типичное использование для SQLDATAReader, и мне было интересно, было ли это возможно, и если так, как заменить цикл при этом с чем-то в TPL. Поскольку читатель не может обеспечить фиксированное количество итераций, которые для расширения метода невозможна, что листья, касающихся задач, которые я собирался. Я надеялся, что кто-то, возможно, решил это уже и разработал некоторые дела и дон с 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

Ты почти там. Оберните код, который вы разместили в функции с этой подписью:

IEnumerable<IDataRecord> MyQuery()

а затем заменить свой // Do something with Reader код с этим:

yield return reader;

Теперь у вас есть то, что работает в одном потоке. К сожалению, когда вы прочитаете результаты запроса, это возвращает ссылку на такой же Объект каждый раз, и объект просто называется для каждой итерации. Это означает, что если вы попытаетесь запустить его параллельно, вы получите некоторые действительно странные результаты, так как параллельно читает, мутируйте объект, используемый в разных потоках. Вам нужен код, чтобы взять скопировать записи, чтобы отправить на вашу параллельную петлю.

На данный момент, однако, что я люблю делать, это пропустить дополнительную копию записи и перейти прямо в класс с сильным типом. Более того, мне нравится использовать общий метод для этого:

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);
            }
        }
    }
}

Предполагая, что ваши фабричные методы создают копию, как и ожидалось, этот код должен быть безопасным для использования в параллельной петле. Вызов метода будет выглядеть примерно так (при условии, что класс сотрудников со статическим фабричным методом под названием «Создание»):

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());

Важное обновление:
Я не такой уверен в этом коде, как я когда-то был. Отдельный поток все еще может мутировать читателя, а другой поток находится в процессе создания его копии. Я мог бы поставить замок вокруг этого, но я также обеспокоен тем, что другой нить может позвонить обновлению читателя после того, как оригинал сам позвал прочитать (), но прежде чем он начнет сделать копию. Следовательно, критический раздел здесь состоит из целого пока цикла ... и на данный момент вы снова вернулись на однопоточную резьбу. Я ожидаю, что есть способ изменить этот код для работы, как ожидалось, для многопоточных сценариев, но ему понадобится больше исследований.

Answer 2

У вас возникли трудности, заменяя, что пока цикла прямо. SQLDATAREADER не является безопасным классом потока, поэтому вы не можете использовать его непосредственно из нескольких потоков.

Что говорится, что вы могли бы потенциально процесс Данные, которые вы читаете с помощью TPL. Здесь есть несколько вариантов. Самый проще всего может сделать свой собственный IEnumerable<T> Реализация, которая работает на читателе и возвращает класс или структуру, содержащую ваши данные. Вы могли бы использовать Plinq или A Parallel.ForEach Заявление для обработки ваших данных параллельно:

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 ...);
            }
        }
    }
}

Как только у вас есть этот метод, вы можете обрабатывать это напрямую, через PLINQ или TPL:

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

Или:

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