Spring data jpa在多线程下获得旧结果

Question

在多线程下，我不断从存储库中获取旧结果。

@Transactional(isolation = Isolation.REPEATABLE_READ)
public void updateScore(int score, Long userId) {
    logger.info(RegularLock.getInstance().getLock().toString());
    synchronized (RegularLock.getInstance().getLock()) {
        Customer customer = customerDao.findOne(userId);
        System.out.println("start:": customer.getScore());
        customer.setScore(customer.getScore().subtract(score));
        customerDao.saveAndFlush(customer);
    }

}

CustomerDao 看起来像

@Transactional
public T saveAndFlush(T model, Long id) {
    T res = repository.saveAndFlush(model);
    EntityManager manager = jpaContext.getEntityManagerByManagedType(model.getClass());
    manager.refresh(manager.find(model.getClass(), id));
    return res;
}

saveAndFlush() from JpaRepository 用于立即保存更改并锁定整个代码。但我仍然不断得到旧的结果。

java.util.concurrent.locks.ReentrantLock@10a9598d[Unlocked]
start:710
java.util.concurrent.locks.ReentrantLock@10a9598d[Unlocked]
start:710

我正在使用带有 spring data jpa 的 springboot。

我把所有代码放在一个测试控制器中，问题依旧

@RestController
@RequestMapping(value = "/test", produces = "application/json")
public class TestController {
    private static Long testId;

    private final CustomerBalanceRepository repository;

    @Autowired
    public TestController(CustomerBalanceRepository repository) {
        this.repository = repository;
    }

    @PostConstruct
    public void init() {
//        CustomerBalance customer = new CustomerBalance();
//        repository.save(customer);
//        testId = customer.getId();
    }


    @SystemControllerLog(description = "updateScore")
    @RequestMapping(method = RequestMethod.GET)
    @Transactional(isolation = Isolation.REPEATABLE_READ)
    public CustomerBalance updateScore() {
        CustomerBalance customerBalance = repository.findOne(70L);
        System.out.println("start:" + customerBalance.getInvestFreezen());
        customerBalance.setInvestFreezen(customerBalance.getInvestFreezen().subtract(new BigDecimal(5)));
        saveAndFlush(customerBalance);
        System.out.println("end:" + customerBalance.getInvestFreezen());
        return customerBalance;
    }

    @Transactional
    public CustomerBalance saveAndFlush(CustomerBalance customerBalance) {
        return repository.saveAndFlush(customerBalance);
    }
}

结果是

start:-110.00
end:-115.00
start:-110.00
end:-115.00
start:-115.00
end:-120.00
start:-120.00
end:-125.00
start:-125.00
end:-130.00
start:-130.00
end:-135.00
start:-130.00
end:-135.00
start:-135.00
end:-140.00
start:-140.00
end:-145.00
start:-145.00
end:-150.00

Answer 1

我试图重现该问题但失败了。我通过请求localhost:8080/test 将您的代码（几乎没有更改）放入控制器并执行它，并且可以在日志中看到分数按预期降低。 注意：它实际上会产生异常，因为我没有配置视图结果，但这应该是无关紧要的。

因此，我建议采取以下行动： 从下面取出我的控制器，将其添加到您的代码中，尽可能少地进行更改。验证它是否确实有效。然后一步步修改，直到和你当前的代码一致。注意开始产生你当前行为的变化。这可能会使原因非常明显。如果没有用您发现的内容更新问题。

@Controller
public class CustomerController {
    private static String testId;

    private final CustomerRepository repository;

    private final JpaContext context;

    public CustomerController(CustomerRepository repository, JpaContext context) {
        this.repository = repository;
        this.context = context;
    }

    @PostConstruct
    public void init() {
        Customer customer = new Customer();
        repository.save(customer);
        testId = customer.id;
    }


    @RequestMapping(path = "/test")
    @Transactional(isolation = Isolation.REPEATABLE_READ)
    public Customer updateScore() {
        Customer customer = repository.findOne(testId);
        System.out.println("start:" + customer.getScore());
        customer.setScore(customer.getScore() - 23);
        saveAndFlush(customer);
        System.out.println("end:" + customer.getScore());
        return customer;
    }

    @Transactional
    public Customer saveAndFlush(Customer customer) {
        return repository.saveAndFlush(customer);
    }
}

经过 OP 的更新和一些讨论，我们似乎已经确定了它：

这个问题只发生在多个线程上（OP 使用 JMeter 来做这件事 10 次/秒）。

Transaction level serializable 似乎也解决了这个问题。

诊断

似乎是更新丢失问题，导致如下效果：

Thread 1: reads the customer score=10 
Thread 2: reads the customer score= 10 
Thread 1: updates the customer to score 10-4 =6
Thread 2: updates the customer to score 10-3 =7 // update from Thread 1 is gone.

为什么不同步？

这里的问题很可能是读取发生在问题中显示的代码之前，因为 EntityManager 是一级缓存。

如何解决 这应该被 JPA 的乐观锁定捕获，因为这需要一个带有 @Version 注释的列。

如果这种情况经常发生，Transaction Level Serializable 可能是更好的选择。

Answer 2

这看起来像悲观锁定的情况。在您的存储库方法中创建 findOneWithLock 像这样：

import org.springframework.data.jpa.repository.Lock;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.JpaRepository;
import org.springframework.data.repository.query.Param;

import javax.persistence.LockModeType;

public interface CustomerRepository extends JpaRepository<Customer, Long> {
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("select c from Customer c where c.id = :id")
    Customer findOneWithLock(@Param("id") long id);
}

并使用它来获取数据库级别的锁，该锁将一直保持到事务结束：

@Transactional
public void updateScore(int score, Long userId) {
    Customer customer = customerDao.findOneWithLock(userId);
    customer.setScore(customer.getScore().subtract(score));
}

无需在代码中使用像 RegularLock 这样的应用程序级锁。

Answer 3

问题似乎是即使你打电话，

customerDao.saveAndFlush(customer);

在到达方法结束并进行提交之前，不会发生提交，因为您的代码位于

@Transactional(isolation = Isolation.REPEATABLE_READ)

您可以做的是将事务的propagation 更改为Propagation.REQUIRES_NEW，如下所示。

@Transactional(isolation = Isolation.REPEATABLE_READ, propagation=Propagation.REQUIRES_NEW)

这将导致在方法结束时创建并提交一个新事务。并在事务结束时提交更改。

Answer 4

用这条线

manager.refresh(manager.find(model.getClass(), id));

您是在告诉 JPA 撤消您的所有更改。来自refresh method的文档

从数据库中刷新实例的状态，覆盖对实体所做的更改（如果有）。

删除它，您的代码应该会按预期运行。