Параллель не работает с Entity Framework

Ответ 1

Базовые подключения к базе данных, которые использует платформа Entity Framework, не являются потокобезопасными. Вам нужно будет создать новый контекст для каждой операции в другом потоке, который вы собираетесь выполнять.

Ваша забота о том, как распараллелить операцию, является действительной; что многие контексты будут дорогими для открытия и закрытия.

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

Если вы можете, создайте новый Task<TResult> (или Task, если вам не нужен результат) для каждой процедуры, а затем в этом Task<TResult>, открыть один контекст, просмотреть все элементы и выполнить хранимую процедуру. Таким образом, у вас есть только несколько контекстов, равных количеству хранимых процедур, которые вы выполняете параллельно.

Предположим, что у вас есть MyDbContext с двумя хранимыми процедурами, DoSomething1 и DoSomething2, оба из которых принимают экземпляр класса MyItem.

Реализация вышеуказанного будет выглядеть примерно так:

// You'd probably want to materialize this into an IList<T> to avoid
// warnings about multiple iterations of an IEnumerable<T>.
// You definitely *don't* want this to be an IQueryable<T>
// returned from a context.
IEnumerable<MyItem> items = ...;

// The first stored procedure is called here.
Task t1 = Task.Run(() => { 
    // Create the context.
    using (var ctx = new MyDbContext())
    // Cycle through each item.
    foreach (MyItem item in items)
    {
        // Call the first stored procedure.
        // You'd of course, have to do something with item here.
        ctx.DoSomething1(item);
    }
});

// The second stored procedure is called here.
Task t2 = Task.Run(() => { 
    // Create the context.
    using (var ctx = new MyDbContext())
    // Cycle through each item.
    foreach (MyItem item in items)
    {
        // Call the first stored procedure.
        // You'd of course, have to do something with item here.
        ctx.DoSomething2(item);
    }
});

// Do something when both of the tasks are done.

Если вы не можете выполнять хранимые процедуры параллельно (каждый из них зависит от запуска в определенном порядке), вы все равно можете распараллелить свои операции, это немного сложнее.

Вы бы посмотрели создание пользовательских разделов по вашим статьям (используя статический Create в Partitioner class). Это даст вам возможность получить IEnumerator<T> реализации (обратите внимание: это не IEnumerable<T>, чтобы вы не могли foreach над ним).

Для каждого экземпляра IEnumerator<T>, который вы возвращаете, вы должны создать новый Task<TResult> (если вам нужен результат), а в теле Task<TResult> вы должны создать контекст, а затем выполнить цикл через возвращаемые элементы с помощью IEnumerator<T>, вызвав хранимые процедуры в порядке.

Это будет выглядеть так:

// Get the partitioner.
OrdinalPartitioner<MyItem> partitioner = Partitioner.Create(items);

// Get the partitions.
// You'll have to set the parameter for the number of partitions here.
// See the link for creating custom partitions for more
// creation strategies.
IList<IEnumerator<MyItem>> paritions = partitioner.GetPartitions(
    Environment.ProcessorCount);

// Create a task for each partition.
Task[] tasks = partitions.Select(p => Task.Run(() => { 
        // Create the context.
        using (var ctx = new MyDbContext())
        // Remember, the IEnumerator<T> implementation
        // might implement IDisposable.
        using (p)
        // While there are items in p.
        while (p.MoveNext())
        {
            // Get the current item.
            MyItem current = p.Current;

            // Call the stored procedures.  Process the item
            ctx.DoSomething1(current);
            ctx.DoSomething2(current);
        }
    })).
    // ToArray is needed (or something to materialize the list) to
    // avoid deferred execution.
    ToArray();

Ответ 2

EF не является потокобезопасным, поэтому вы не можете использовать Parallel.

Взгляните на Entity Framework и многопоточность

и статья.

Ответ 3

Это то, что я использую и прекрасно работаю. Он также поддерживает обработку исключений ошибок и имеет режим отладки, который значительно упрощает отслеживание событий.

public static ConcurrentQueue<Exception> Parallel<T>(this IEnumerable<T> items, Action<T> action, int? parallelCount = null, bool debugMode = false)
{
    var exceptions = new ConcurrentQueue<Exception>();
    if (debugMode)
    {
        foreach (var item in items)
        {
            try
            {
                action(item);
            }
            // Store the exception and continue with the loop.                     
            catch (Exception e)
            {
                exceptions.Enqueue(e);
            }
        }
    }
    else
    {
        var partitions = Partitioner.Create(items).GetPartitions(parallelCount ?? Environment.ProcessorCount).Select(partition => Task.Factory.StartNew(() =>
        {
            while (partition.MoveNext())
            {
                try
                {
                    action(partition.Current);
                }
                // Store the exception and continue with the loop.                     
                catch (Exception e)
                {
                    exceptions.Enqueue(e);
                }
            }
        }));
        Task.WaitAll(partitions.ToArray());
    }
    return exceptions;
}

Вы используете его, как показано ниже, где db - это оригинальный DbContext, а db.CreateInstance() создает новый экземпляр с использованием той же строки соединения.

        var batch = db.Set<SomeListToIterate>().ToList();
        var exceptions = batch.Parallel((item) =>
        {
            using (var batchDb = db.CreateInstance())
            {
                var batchTime = batchDb.GetDBTime();
                var someData = batchDb.Set<Permission>().Where(x=>x.ID = item.ID).ToList();
                //do stuff to someData
                item.WasMigrated = true; //note that this record is attached to db not batchDb and will only be saved when db.SaveChanges() is called
                batchDb.SaveChanges();        
            }                
        });
        if (exceptions.Count > 0)
        {
            logger.Error("ContactRecordMigration : Content: Error processing one or more records", new AggregateException(exceptions));
            throw new AggregateException(exceptions); //optionally throw an exception
        }
        db.SaveChanges(); //save the item modifications

Ответ 4

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

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