Как правильно обрабатывать два потока, обновляющих одну и ту же строку в базе данных

Ответ 1

Часть 1 - Ваша проблема - то, как я ее вижу.

Основной причиной получения этого исключения является то, что вы используете Hibernate с возможно оптимистичной блокировкой. Это в основном говорит вам, что либо поток T1, либо поток T2 уже обновили состояние до PARSED, а теперь в другом потоке находится старая версия строки с меньшей версией, чем та, которая хранится в базе данных, и пытается обновить состояние до PARSED.

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

Основная проблема возникает, если после того, как состояние установлено в RECIEVED, если два потока T1 и T2 фактически зависят друг от друга при переходе к следующему статусу. В этом случае вам нужно убедиться, что если T1 выполнил первый (или наоборот), T2 должен обновить данные для обновленной строки и повторно применить свои изменения на основе изменений, уже введенных T1. В этом случае решение является следующим. Если вы столкнулись с staleObjectException, вам в основном нужно обновить данные из базы данных и перезапустить свою работу.

Анализ части 2 по размещенной ссылке Возможные исключения в спящем режиме, когда два потока обновляют один и тот же объект? Подход 1, это более или менее последняя версия для обновления Wins. Он более или менее избегает оптимистической блокировки (подсчет версии). Если у вас нет зависимости от T1 до T2 или наоборот, чтобы установить статус PARSED. Это должно быть хорошо.

**** Aproach 2 ** Оптимистичное блокирование ** Это то, что у вас есть сейчас. Решение состоит в том, чтобы обновить данные и перезапустить вашу работу.

Aproach 3 Блокировка уровня уровня строки. Решение здесь более или менее такое же, как для подхода 2 с небольшой коррекцией, которую придерживается пессимистический замок. Основное различие заключается в том, что в этом случае это может быть блокировка READ, и вы даже не сможете читать данные из базы данных, чтобы обновить ее, если она PESSIMISTIC READ.

Синхронизация уровня приложения Aproach 4 Существует много разных способов синхронизации. Одним из примеров было бы фактически организовать все ваши обновления в очереди BlockingQueue или JMS (если вы хотите, чтобы она была постоянной) и выталкивать все обновления из одного потока. Чтобы визуализировать это, бит T1 и T2 будет помещать элементы в очередь, и будут выполняться отдельные операции чтения потоков T3 и нажимать их на сервер базы данных.

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

Ну, я пока не могу придумать ничего другого:)

Ответ 2

Я не уверен, что я понимаю вопрос, но, похоже, это будет логической ошибкой для потока T1, который обрабатывает, например, записи, начинающиеся с AA, чтобы пометить весь файл как "Parsed"? Что произойдет, если, например, ваше приложение аварийно завершает работу после обновлений T1, но пока T2 все еще обрабатывает записи BB? Некоторые записи BB, вероятно, будут потеряны, правильно?

Во всяком случае, суть проблемы в том, что у вас есть условие гонки с двумя потоками, обновляющими один и тот же объект. Исключение устаревшего объекта означает, что один из ваших потоков потерял гонку. Лучшее решение полностью избегает гонки.

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

Я бы использовал функции java.util.concurrent для отправки записей в потоковые рабочие и имел поток, взаимодействующий с блоком hibernate, до тех пор, пока все записи не будут обработаны, и в этот момент этот поток может пометить файл как "Parsed".

Например,

// do something like this during initialization, or use a Guava LoadingCache...
Map<RecordType, Executor> executors = new HashMap<>();
// note I'm assuming RecordType looks like an enum
executors.put(RecordType.AA_RECORD, Executors.newSingleThreadExecutor());

то при обработке файла вы отправляете каждую запись следующим образом, создавая список фьючерсов, соответствующих статусу поставленных задач. Предположим, что успешная обработка записи возвращает логическое значение "true":

List<Future<Boolean>> tasks = new ArrayList<>();
for (Record record: file.getRecords()) {
    Executor executorForRecord = executors.get(record.getRecordType());
    tasks.add(executor.submit(new RecordProcessor(record)));
}

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

boolean allSuccess = true;
for (Future<Boolean> task: tasks) {
    allSuccess = allSuccess && task.get();
    if (!allSuccess) break;
}

// if all your tasks completed successfully, update the file record
if (allSuccess) {
    file.setStatus("Parsed");
}

Ответ 3

Предполагая, что каждый поток T1, T2 будет анализировать разные части файла, означает, что никто не переопределяет другой синтаксический анализ потока. лучше всего отделить ваш процесс синтаксического анализа от фиксации БД.

T1, T2 выполнит разбор T3 или Main Thread выполнит фиксацию после завершения T1, T2. и я думаю, что при таком подходе более корректно изменить статус файла на Parsed только тогда, когда оба потока завершены.

вы можете думать о T3 как о классе CommitService, который ждет до тех пор, пока T1, T2 finsih и не зафиксирует DB

CountDownLatch - полезный инструмент для этого. и здесь Example