休眠会话未在线程之间共享

Question

我有一个实现用户推荐系统的 springboot 应用程序。一个用例是，当用户使用另一个用户的有效推荐代码注册时，推荐用户获得一个奖励积分，每 5 个积分他们获得 10 美元的信用。据此，我在我的应用程序中实现了一个遵守这些业务规则的用例，为了测试高并发下的正确行为，我使用@DataJpaTest 和 spring 数据存储库和 H2 DB 作为存储系统创建了一个集成测试。在我的测试中，我创建了第一个用户，然后使用第一个用户推荐代码创建了一定数量的用户，这些用户中的每一个都是使用线程池执行程序在不同的线程上创建的，以产生这些线程。我的问题是通过线程池衍生线程创建的用户看不到在主线程中创建的第一个用户，即使我使用JpaRepository.saveAndFlush() 方法来保存它们。

有人能给我解释一下这里发生了什么吗？是不是因为 Hibernate 的 session 不是线程安全的？

您可以在下面看到我的代码，第一个测试已简化为仅检查存储库中的用户数量。

@DataJpaTest(includeFilters = @ComponentScan.Filter(type = FilterType.ANNOTATION, classes = Repository.class))
public class JpaTest {

    @Autowired
    private TestEntityManager entityManager;

    @Autowired
    @Qualifier("JpaUserRepository")
    private JpaUserRepository userRepository;

    @Autowired
    @Qualifier("JpaReferralRepository")
    private ReferralRepository referralRepository;

    private RegisterReferredUser registerReferredUser;
    private CreateUser createUser;
    private GetUser getUser;

    @BeforeEach
    void setUp() {
        registerReferredUser = new RegisterReferredUser(referralRepository, userRepository);
        createUser = new CreateUser(userRepository, referralRepository, registerReferredUser);
        getUser = new GetUser(userRepository);
    }

    @Test
    void createUser_shouldWorksProperlyUnderConcurrentExecution() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        EmailAddress referrerUserEmail = EmailAddress.of("john.doe@acme.inc");
        User referrerUser = createUser.execute(new CreateUserCommand(referrerUserEmail.getValue(), null));
        String referralCode = referrerUser.getReferralCode().getValue();
        int maxIterations = 10;

        for (int i = 0; i < maxIterations; i++) {
            int emailSeed = i;
            executor.submit(() -> {
                    createUser.execute(new CreateUserCommand(anEmailAddress(emailSeed), referralCode));
            });
        }

        executor.shutdown();
        if (!executor.awaitTermination(20, TimeUnit.SECONDS)) {
            fail("Executor didn't finish in time");
        }

        assertThat(entityManager.getEntityManager().createQuery("from JpaUser").getResultList().size()).isEqualTo(maxIterations + 1);
        // This fails: just 1 user in the repository, however, if I register users without referral (without checking the existence of the first user), users are created and this pass
    }

    @Test
    void just_a_POC() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        userRepository.save(UserMother.aUserWithEmail("john.doe@acme.inc"));
        int maxIterations = 10;

        for (int i = 0; i < maxIterations; i++) {
            int emailSeed = i;
            executor.submit(() -> userRepository.save(UserMother.aUserWithEmail(anEmailAddress(emailSeed))));
        }

        executor.shutdown();
        if (!executor.awaitTermination(20, TimeUnit.SECONDS)) {
            fail("Executor didn't finish in time");
        }

        assertThat(entityManager.getEntityManager().createQuery("from JpaUser").getResultList().size()).isEqualTo(maxIterations + 1);
        // This pass
    }
}

在CreateUser我有以下代码：

private void assertReferralCodeIsValid(ReferralCode referralCode, EmailAddress email) {
    if (!userRepository.findByReferralCode(referralCode).isPresent()) {
        throw new NonExistentReferralCode(referralCode);
    }
    if (referralRepository.findByEmailAndCode(email, referralCode).isPresent()) {
        throw new ReferralCodeAlreadyUsed(email, referralCode);
    }
}

这就是JpaUserRepository.save() 方法：

@Repository("JpaUserRepository")
public class JpaUserRepository implements UserRepository {
    
    private JpaUserCrudRepository crudRepository;

    public JpaUserRepository(JpaUserCrudRepository crudRepository) {
        this.crudRepository = crudRepository;
    }
       
    @Override
    public void save(User user) {
        crudRepository.saveAndFlush(JpaUser.fromDomain(user));
    }

}

Answer 1

查看为您的事务配置的隔离级别。数据库引擎通常会尝试尽可能快地提供数据而不会阻塞（如果可能）。因此，如果您的所有线程同时读取一个表，它们可能会得到一个“未提交”版本的记录。

如果需要同步，可以更改隔离级别，或在处理表之前锁定表。

更多关于这个话题：

• Spring @Transactional - isolation, propagation
• https://www.baeldung.com/java-jpa-transaction-locks