Полезные советы по использованию EF в многопоточной программе?

Question

У вас есть хорошие советы по использованию EF в многопоточной программе?

У меня есть 2 слоя:

• слой EF для чтения / записи в мою базу данных
• многопоточный сервис, который использует мои сущности (чтение / запись) и выполняет некоторые вычисления (я использую Task Parallel Library в рамках)

Как я могу синхронизировать контексты моего объекта в каждом потоке? Знаете ли вы хороший шаблон, чтобы заставить его работать?

Answer 1

Хороший совет - просто не надо: -) EF едва удается выжить в одной нити - природе зверя.

Если вам абсолютно необходимо его использовать, сделайте самые легкие DTO-ы, закройте OC, как только у вас будут данные, перепакуйте данные, порождайте ваши потоки просто для выполнения вычислений и ничего больше, дождитесь их завершения, затем создайте другой OC и сбросить данные обратно в БД, согласовать их и т. д.

Если другой «основной» поток (тот, который порождает N потоков вычислений через TPL) должен знать, когда какой-то поток завершается, запускается событие, просто установите флаг в другом потоке, а затем позвольте его коду проверить флаг в своем Зацикливайтесь и реагируйте, создавая новый OC, а затем сверяйте данные, если это необходимо.

Если ваша ситуация более проста, вы можете адаптировать ее - ключ в том, что вы можете только установить флаг и позволить другому потоку реагировать, когда он будет готов. Это означает, что он находится в стабильном состоянии, завершил раунд того, что он делал, и может делать вещи, не рискуя условиями гонки. Сбросьте флаг (целое число) с помощью взаимосвязанных операций и сохраните некоторые временные данные, чтобы убедиться, что ваши потоки не реагируют снова в течение некоторого времени T - в противном случае они могут потратить свою жизнь, просто запрашивая БД.

Answer 2

Так я реализовал свой сценарий.

var processing= new ConcurrentQueue<int>();


//possible multi threaded enumeration only processed non-queued records
Parallel.ForEach(dataEnumeration, dataItem=>
{
     if(!processing.Contains(dataItem.Id))
     {
         processing.Enqueue(dataItem.Id);

          var myEntityResource = new EntityResource();

          myEntityResource.EntityRecords.Add(new EntityRecord
                                      {
                                        Field1="Value1", 
                                        Field2="Value2"
                                      }
                               );

           SaveContext(myEntityResource);

       var itemIdProcessed = 0;
       processing.TryDequeue(out itemIdProcessed );

     }

}

public void RefreshContext(DbContext context)
    {
        var modifiedEntries = context.ChangeTracker.Entries()
            .Where(e => e.State == EntityState.Modified || e.State == EntityState.Deleted);
        foreach (var modifiedEntry in modifiedEntries)
        {
            modifiedEntry.Reload();
        }
    }

public bool SaveContext(DbContext context,out Exception error, bool reloadContextFirst = true)
    {
        error = null;
        var saved = false;
        try
        {
            if (reloadContextFirst)
                this.RefreshContext(context);
            context.SaveChanges();
            saved = true;
        }
        catch (OptimisticConcurrencyException)
        {
            //retry saving on concurrency error
            if (reloadContextFirst)
                this.RefreshContext(context);
            context.SaveChanges();
            saved = true;
        }
        catch (DbEntityValidationException dbValEx)
        {
            var outputLines = new StringBuilder();
            foreach (var eve in dbValEx.EntityValidationErrors)
            {
                outputLines.AppendFormat("{0}: Entity of type \"{1}\" in state \"{2}\" has the following validation errors:",
                    DateTime.Now, eve.Entry.Entity.GetType().Name, eve.Entry.State);
                foreach (var ve in eve.ValidationErrors)
                {
                    outputLines.AppendFormat("- Property: \"{0}\", Error: \"{1}\"", ve.PropertyName, ve.ErrorMessage);
                }
            }

            throw new DbEntityValidationException(string.Format("Validation errors\r\n{0}", outputLines.ToString()), dbValEx);
        }
        catch (Exception ex)
        {
            error = new Exception("Error saving changes to the database.", ex);
        }
        return saved;
    }

Answer 3

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

Это не идеально.Ваш слой базы данных будет эффективно однопоточным.Но он сохранил многопоточность в многопоточной среде.В моем случае, тяжелые вычисления вообще отсутствовали в коде базы данных - это был игровой сервер, поэтому игровая логика была, конечно же, основным ресурсом.Поэтому у меня не было особой необходимости в многопоточном слое БД.

Answer 4

Я думаю, что ваш вопрос больше о синхронизации между потоками и EF здесь не имеет значения.Если я правильно понимаю, вы хотите уведомить потоки из одной группы, когда основной поток выполнил какую-либо операцию - в данном случае операцию «SaveChanges ()».Потоки здесь похожи на клиент-серверные приложения, где один поток является сервером, а другие потоки являются клиентами, и вы хотите, чтобы клиентские потоки реагировали на активность сервера.

Как кто-то заметил, вам, вероятно, не нужны потоки, но давайте оставим все как есть.

Нет страха перед мертвыми блокировками, если вы собираетесь использовать отдельный OC для потока.

Я также предполагаю, что ваши клиентские потоки являются долгосрочными потоками в некотором цикле.Если вы хотите, чтобы ваш код выполнялся в клиентском потоке, вы не можете использовать события C #.

class ClientThread {
public bool SomethingHasChanged;

  public MainLoop()
  {
    Loop {
      if (SomethingHasChanged)
      { 
        refresh();
        SomethingHasChanged = false;
      }

      // your business logic here


    } // End Loop
  }
}

Теперь вопрос в том, как вы установите флаг во всех ваших клиентских потоках?Вы можете хранить ссылки на клиентские потоки в вашем основном потоке, проходить через них и устанавливать для всех флагов значение true.

Answer 5

"Как я могу синхронизировать контексты моего объекта в каждом потоке?"Это будет сложно.Прежде всего, запросы SP или DB могут иметь план параллельного выполнения.Поэтому, если у вас также есть параллелизм в контексте объекта, вы должны вручную убедиться, что у вас есть достаточная изоляция, но достаточно, чтобы вы не удерживали блокировку слишком долго, чтобы вызвать взаимные блокировки.

Так что я бы сказал, что делать это не нужно.

Но это может быть не тот ответ, который вам нужен.Можете ли вы объяснить немного больше, чего вы хотите достичь, используя эту многопоточность.Является ли он более вычислительным или IO-связанным.Если это долгосрочные операции, связанные с IO, то посмотрите на APM Джеффа Рихтера.

Answer 6

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

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

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

Answer 7

Я думаю, что Крейг может быть прав насчет того, что вашему приложению не нужны потоки ... но вы можете искать использование ConcurrencyCheck в своих моделях, чтобы убедиться, что вы не "переопределяете" свои изменения