Автор оригинала: Vlad Mihalcea.
В моем предыдущем посте я объяснил преимущества использования явной оптимистической блокировки . Как мы затем обнаружили, существует очень короткое временное окно, в течение которого параллельная транзакция все еще может зафиксировать изменение цены продукта прямо перед тем, как будет зафиксирована наша текущая транзакция.
Эту проблему можно описать следующим образом:
- Алиса приносит товар
- Затем она решает заказать его
- Приобретена блокировка оптимистичного продукта
- Заказ вставляется в текущий сеанс базы данных транзакций
- Версия продукта проверяется с помощью процедуры явной оптимистической блокировки Hibernate
- Механизм ценообразования управляет фиксацией изменения цены продукта
- Транзакция Alice совершается без осознания того, что цена продукта только что изменилась
Поэтому нам нужен способ ввести изменение цены продукта между оптимистичной проверкой блокировки и фиксацией транзакции заказа.
После анализа исходного кода Hibernate мы обнаруживаем, что метод SessionImpl.beforeTransactionCompletion() вызывает текущий настроенный Перехватчик.beforeTransactionCompletion() обратный вызов сразу после внутреннего actionQueue обработчика этапа (где проверяется явная оптимистичная версия заблокированной сущности):
public void beforeTransactionCompletion(TransactionImplementor hibernateTransaction) {
LOG.trace( "before transaction completion" );
actionQueue.beforeTransactionCompletion();
try {
interceptor.beforeTransactionCompletion( hibernateTransaction );
}
catch (Throwable t) {
LOG.exceptionInBeforeTransactionCompletionInterceptor( t );
}
}
Вооружившись этой информацией, мы можем настроить тест, чтобы воспроизвести наше состояние гонки:
private AtomicBoolean ready = new AtomicBoolean();
private final CountDownLatch endLatch = new CountDownLatch(1);
@Override
protected Interceptor interceptor() {
return new EmptyInterceptor() {
@Override
public void beforeTransactionCompletion(Transaction tx) {
if(ready.get()) {
LOGGER.info("Overwrite product price asynchronously");
executeAsync(() -> {
Session _session = getSessionFactory().openSession();
_session.doWork(connection -> {
try (PreparedStatement ps = connection.prepareStatement("UPDATE product set price = 14.49 WHERE id = 1")) {
ps.executeUpdate();
}
});
_session.close();
endLatch.countDown();
});
try {
LOGGER.info("Wait 500 ms for lock to be acquired!");
Thread.sleep(500);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
}
}
};
}
@Test
public void testExplicitOptimisticLocking() throws InterruptedException {
try {
doInTransaction(session -> {
try {
final Product product = (Product) session.get(Product.class, 1L, new LockOptions(LockMode.OPTIMISTIC));
OrderLine orderLine = new OrderLine(product);
session.persist(orderLine);
lockUpgrade(session, product);
ready.set(true);
} catch (Exception e) {
throw new IllegalStateException(e);
}
});
} catch (OptimisticEntityLockException expected) {
LOGGER.info("Failure: ", expected);
}
endLatch.await();
}
protected void lockUpgrade(Session session, Product product) {}
При его запуске тест генерирует следующие выходные данные:
#Alice selects a Product
DEBUG [main]: Query:{[select abstractlo0_.id as id1_1_0_, abstractlo0_.description as descript2_1_0_, abstractlo0_.price as price3_1_0_, abstractlo0_.version as version4_1_0_ from product abstractlo0_ where abstractlo0_.id=?][1]}
#Alice inserts an OrderLine
DEBUG [main]: Query:{[insert into order_line (id, product_id, unitPrice, version) values (default, ?, ?, ?)][1,12.99,0]}
#Alice transaction verifies the Product version
DEBUG [main]: Query:{[select version from product where id =?][1]}
#The price engine thread is started
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Overwrite product price asynchronously
#Alice thread sleeps for 500ms to give the price engine a chance to execute its transaction
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Wait 500 ms for lock to be acquired!
#The price engine changes the Product price
DEBUG [pool-1-thread-1]: Query:{[UPDATE product set price = 14.49 WHERE id = 1][]}
#Alice transaction is committed without realizing the Product price change
DEBUG [main]: o.h.e.t.i.j.JdbcTransaction - committed JDBC Connection
Итак, условия гонки реальны. Вам решать, требует ли ваше текущее приложение более строгих требований к целостности данных, но, как правило, лучше перестраховаться, чем сожалеть.
Чтобы устранить эту проблему, нам просто нужно добавить пессимистичный запрос на блокировку непосредственно перед завершением нашего транзакционного метода.
@Override
protected void lockUpgrade(Session session, Product product) {
session.buildLockRequest(new LockOptions(LockMode.PESSIMISTIC_READ)).lock(product);
}
Явная общая блокировка предотвратит одновременную запись на объект, который мы ранее оптимистично заблокировали. С помощью этого метода никакая другая параллельная транзакция не может изменить Продукт до снятия этой блокировки (после фиксации или отката текущей транзакции).
При наличии нового пессимистического запроса на блокировку предыдущий тест выдает следующий результат:
#Alice selects a Product
DEBUG [main]: Query:{[select abstractlo0_.id as id1_1_0_, abstractlo0_.description as descript2_1_0_, abstractlo0_.price as price3_1_0_, abstractlo0_.version as version4_1_0_ from product abstractlo0_ where abstractlo0_.id=?][1]}
#Alice inserts an OrderLine
DEBUG [main]: Query:{[insert into order_line (id, product_id, unitPrice, version) values (default, ?, ?, ?)][1,12.99,0]}
#Alice applies an explicit physical lock on the Product entity
DEBUG [main]: Query:{[select id from product where id =? and version =? for update][1,0]}
#Alice transaction verifies the Product version
DEBUG [main]: Query:{[select version from product where id =?][1]}
#The price engine thread is started
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Overwrite product price asynchronously
#Alice thread sleeps for 500ms to give the price engine a chance to execute its transaction
INFO [main]: c.v.h.m.l.c.LockModeOptimisticRaceConditionTest - Wait 500 ms for lock to be acquired!
#The price engine cannot proceed because of the Product entity was locked exclusively, so Alice transaction is committed against the ordered Product price
DEBUG [main]: o.h.e.t.i.j.JdbcTransaction - committed JDBC Connection
#The physical lock is released and the price engine can change the Product price
DEBUG [pool-1-thread-1]: Query:{[UPDATE product set price = 14.49 WHERE id = 1][]}
Несмотря на то, что мы запросили блокировку PESSIMISTIC_READ , вместо этого HSQLDB может выполнить только блокировку ДЛЯ ОБНОВЛЕНИЯ, эквивалентную явному режиму PESSIMISTIC_WRITE блокировки.
Если вы задаетесь вопросом, почему мы используем как оптимистическую, так и пессимистическую блокировку для нашей текущей транзакции, вы должны помнить, что оптимистическая блокировка является единственным возможным механизмом управления параллелизмом для разговоров по нескольким запросам .
В нашем примере сущность Продукта загружается по первому запросу с использованием транзакции, доступной только для чтения. Сущность продукта имеет связанную версию, и этот объект entitysnapshot, основанный на чтении, будет оптимистично заблокирован во время транзакции во время записи.
Пессимистическая блокировка полезна только во время транзакции во время записи, чтобы предотвратить одновременное обновление после проверки версии сущности продукта. Таким образом, как логическая блокировка, так и физическая блокировка взаимодействуют для обеспечения целостности данных о цене заказа.
Пока я работал над этим сообщением в блоге, Чемпион Java Маркус Айзеле взял у меня интервью об Мастер-классе Hibernate инициатива. Во время интервью я попытался объяснить примеры текущих сообщений, подчеркнув при этом истинную важность знания ваших инструментов за пределами справочной документации.
Код доступен на GitHub .