ConcurrentTest.java

ConcurrentTest.java

package com.benx.weply.core.spring.jpa;
import com.benx.weply.core.SpringTest;
import com.benx.weply.core.domain.enums.Shop;
import com.benx.weply.core.domain.user.User;
import com.benx.weply.core.domain.user.UserService;
import lombok.extern.slf4j.Slf4j;
import org.hibernate.StaleObjectStateException;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.EntityTransaction;
import java.util.concurrent.CountDownLatch;
@Slf4j
public class ConcurrentTest extends SpringTest {
    @Autowired
    private EntityManagerFactory entityManagerFactory;
    @Autowired
    private UserService userService;
    private final CountDownLatch loadProductsLatch = new CountDownLatch(3);
    private final CountDownLatch latch1 = new CountDownLatch(1);
    private final CountDownLatch latch2 = new CountDownLatch(1);
    private final CountDownLatch latch3 = new CountDownLatch(1);
    @Test
    public void update() throws InterruptedException {
        User actual = userService.findById(1L);
        log.info("actual : {}, {}", actual, actual.getFirstName());
        Thread one = new Thread(new OneTransaction());
        Thread two = new Thread(new TwoTransaction());
        Thread three = new Thread(new ThreeTransaction());
        one.start();
        two.start();
        three.start();
        one.join();
        two.join();
        three.join();
    }
    public class OneTransaction implements Runnable {
        @Override
        public void run() {
            try {
                doInTransaction(new TransactionCallable() {
                    @Override
                    public Object execute(EntityManager entityManager) {
                        try {
                            User user = entityManager.find(User.class, 1L);
                            loadProductsLatch.countDown();
                            loadProductsLatch.await();
                            log.info("LATCH1 FIRED");
                            user.setFirstName("111");
                            return user;
                        } catch (InterruptedException e) {
                            throw new IllegalStateException(e);
                        }
                    }
                });
            } catch (StaleObjectStateException expected) {
                log.info("One: Optimistic locking failure", expected);
            }
            log.info("LATCH1 DONE");
            latch1.countDown();
        }
    }
    public class TwoTransaction implements Runnable {
        @Override
        public void run() {
            try {
                doInTransaction(new TransactionCallable() {
                    @Override
                    public Object execute(EntityManager entityManager) {
                        try {
                            User user = (User) entityManager.find(User.class, 1L);
                            loadProductsLatch.countDown();
                            loadProductsLatch.await();
                            log.info("LATCH2 FIRED");
                            user.setFirstName("222");
                            return user;
                        } catch (InterruptedException e) {
                            throw new IllegalStateException(e);
                        }
                    }
                });
            } catch (StaleObjectStateException expected) {
                log.info("Two: Optimistic locking failure", expected);
            }
            log.info("LATCH1 DONE");
            latch2.countDown();
        }
    }
    public class ThreeTransaction implements Runnable {
        @Override
        public void run() {
            try {
                doInTransaction(new TransactionCallable() {
                    @Override
                    public Object execute(EntityManager entityManager) {
                        try {
                            User user = (User) entityManager.find(User.class, 1L);
                            loadProductsLatch.countDown();
                            loadProductsLatch.await();
                            log.info("LATCH3 FIRED");
                            user.setFirstName("333");
                            return user;
                        } catch (InterruptedException e) {
                            throw new IllegalStateException(e);
                        }
                    }
                });
            } catch (StaleObjectStateException expected) {
                log.info("Three: Optimistic locking failure", expected);
            }
            log.info("LATCH3 FIRED");
            latch3.countDown();
        }
    }
    protected static abstract class TransactionCallable<T> {
        public abstract T execute(EntityManager entityManager);
    }
    protected <T> T doInTransaction(TransactionCallable<T> callable) {
        T result = null;
        EntityManager entityManager = null;
        EntityTransaction txn = null;
        try {
            entityManager = entityManagerFactory.createEntityManager();
            txn = entityManager.getTransaction();
            txn.begin();
            result = callable.execute(entityManager);
            txn.commit();
        } catch (RuntimeException e) {
            if ( txn != null && txn.isActive() ) txn.rollback();
            throw e;
        } finally {
            if (entityManager != null) {
                entityManager.close();
            }
        }
        return result;
    }
}