IsolationLevel.RepeatableRead для предотвращения дублирования

Question

Я работаю над приложением, которое должно создавать продукты (например, страховые полисы) при получении уведомлений о мгновенных платежах PayPal. К сожалению, PayPal иногда отправляет дубликаты уведомлений. Кроме того, есть еще одна третья сторона, которая одновременно выполняет обновления веб-служб, когда они также получают обновления от PayPal.

Вот базовая схема таблиц базы данных.

// table "package"
// columns packageID, policyID, other data...
// 
// table "insurancepolicy"
// columns policyID, coverageAmount, other data...

Вот базовая схема того, что я хочу сделать:

using (SqlConnection conn = new SqlConnection(...))
{
  sqlTransaction sqlTrans = conn.BeginTransaction(IsolationLevel.RepeatableRead);

  // Calls a stored procedure that checks if the foreign key in the transaction table has a value.
  if (PackageDB.HasInsurancePolicy(packageID, conn))
  { 
    sqlTrans.Commit();
    return false;
  }

  // Insert row in foreign table.
  int policyID = InsurancePolicyDB.Insert(coverageAmount, conn);
  if (policyID <= 0)
  {
    sqlTrans.Rollback();
    return false;
  }

  // Assign foreign key to parent table.  If this fails, roll back everything.
  bool assigned = PackageDB.AssignPolicyID(packageID, policyID, conn);
  if (!assigned)
  {
    sqlTrans.Rollback();
    return false;
  }
}

Если два (или более) потока (или процессы или приложения) делают это одновременно, я хочу, чтобы первый поток заблокировал строку «пакет», пока у нее нет идентификатора политики, до тех пор, пока не будет создана политика и policyID назначается таблице пакетов. Тогда блокировка будет снята после того, как policyID назначен таблице пакетов. Я надеюсь, что другой поток, вызывающий этот же код, остановится, когда прочитает строку пакета, чтобы убедиться, что у него сначала нет идентификатора policyID. Когда будет снята блокировка первой транзакции, я надеюсь, что вторая транзакция увидит, что существует policyID, и, следовательно, вернется без вставки каких-либо строк в таблицу политик.

Примечание. Из-за структуры базы данных CRUD каждая хранимая процедура включала либо чтение (выбор), создание (вставку) или обновление.

Правильно ли используется изоляция транзакции RepeatableRead?

Спасибо.

Answer 1

Было бы безопаснее и чище, если бы insert into Policy просто попал в какое-то ограничение таблицы уникальности при попытке вставить дубликат. Повышение уровня изоляции может снизить параллелизм и привести к другим неприятным проблемам, таким как тупики.

Другой способ - всегда вставлять строку политики, а затем откатывать ее, если пакет уже был присоединен к политике:

begin tran (read committed)

/* tentatively insert new Policy */
insert Policy

/* attach Package to Policy if it's still free */
update Package
  set Package.policy_id = @policy_id
  where Package.package_id = @package_id and Package.policy_id is null

if @@rowcount > 0
  commit
else
  rollback

Это лучше всего работает, когда конфликты редки, что, кажется, ваш случай.

Answer 2

Я полагаю, что вы на самом деле хотите Serializable уровень изоляции. Проблема в том, что два потока могут пройти через HasInsurancePolicyCheck (хотя я понятия не имею, что сделает InsurancePolicyDB.Insert или почему он вернет 0)

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

Answer 3

Я согласен с идеей "очереди сообщений" в ответе Аароненсена. Если вас беспокоит то, что несколько одновременно работающих потоков пытаются обновить одну и ту же строку данных одновременно, вам следует вместо этого добавить потоки в свои рабочие очереди, которые затем последовательно обрабатываются одним потоком. Это значительно снижает конкуренцию в базе данных, поскольку целевая таблица обновляется только одним потоком вместо «N», а операции рабочей очереди ограничиваются вставками потоков обмена сообщениями и чтением / обновлением потоком обработки данных.