多线程案例 | 多线程共享同一个事务，并发保存

1. 背景

项目中有一个核心下单业务，每次在保存的时候要同时处理20种其他关联资源的相关业务，并保存。而这20种资源在逻辑上是可以并行保存的，因此想到使用多线程(代码1-1)。如果这些资源的保存在再多个线程，同一个事务内，一个保存失败，整体回滚，不改过原有业务，并发执行20种资源相关业务，这一优化将至少减少40%的响应时间，提高执行速率。
遗憾的是，无论mybatis还是hibernate事务默认是不能多线程共享的。百度了一下，还是没找到满意的解决方案。于是自己写了个demo，实现多个线程使用同一个session. 最后等待所有线程执行完毕，整体提交。

public Response doSaveAll() {
        
        this.doCheck();
    
        //导
        this.doSaveGuiderList();
            //车
        this.doSaveCarList();

              ..... 等等其他近20种资源
        
        //处理日志
        this.doHandleLogAndStat();
        this.doSaveFinally();
        return Response.success("保存成功").addData(cisTrip);
    }

2.DEMO实现

2.1 Employee实体

Employee 员工实体，hebinate entity
EmployeeUtil.getRanE(); //生成一个随机id,name和其他信息的员工

2.2 自定义hibernate的SessionContext --> 继承自SpringSessionContext

2.2.1 SessionContext的currentSession

每次执行sql,HibernateSessionFactory.getCurrentSession()会从SessionContext中取currentSession,而Spring使用SpringSessionContext实现SessionContext。

public Session getCurrentSession() throws HibernateException {
        if ( currentSessionContext == null ) {
            throw new HibernateException( "No CurrentSessionContext configured!" );
        }
      return currentSessionContext.currentSession();
}

2.2.2 自定义的SpringSessionContext --> HwCurrentSessionContext

实现原理 : 每次执行sql,如果用户调用 HwCurrentSessionContext.setTlCurrentSession为当前线程设置了Sessioin(存放在ThreadLocal)，则直接从threadLocal中取，如果用户没有自定义当前线程的session,则使用SpringSessionContext中的session.

@Data
public class TlSessionBean implements Serializable {

    @Description("是否为线程自定义session")
    private Boolean custom;
    private Session session;
}

public class HwCurrentSessionContext extends SpringSessionContext {
    public static final ThreadLocal<TlSessionBean> sessions = new ThreadLocal<>();
    private static      SessionFactory             sessionFactory;

    public HwCurrentSessionContext(SessionFactoryImplementor factory) {
        super(factory);
    }

    public static SessionFactory getSessionFactory() {
        if (sessionFactory == null) {
            synchronized (HibernateUtil.class) {
                if (sessionFactory == null) sessionFactory = BeanContext.getBean(SessionFactory.class);
            }
        }
        return sessionFactory;
    }

    @Override
    public Session currentSession() throws HibernateException {
        TlSessionBean tlSessionBean = sessions.get();
        if (tlSessionBean != null && tlSessionBean.getCustom()) {
            // todo: 如果ThreadLocal存在session，并且是用户自定义session且返回这个session,否则使用springSessionContext的session
            return tlSessionBean.getSession();
        }
        return super.currentSession();
    }

    @Description("获取当前session")
    static Session getCurrentSession() {
        Session session = getTlCurrentSession();
        if (session ==null) session = getSessionFactory().getCurrentSession();
        return session;
    }
    @Description("为线程在ThreadLocal中自定义session")
    static void setTlCurrentSession(Session session) {
        if (getTlCurrentSession() == null) {
            TlSessionBean sessionBean = new TlSessionBean();
            sessionBean.setCustom(true);
            sessionBean.setSession(session);
            sessions.set(sessionBean);
        }
    }
    @Description("获取线程在ThreadLocal中自定义session，没有返回Null")
    static Session getTlCurrentSession() {
        //通过线程对象.get()方法安全创建Session
        TlSessionBean s = sessions.get();
        // 如果该线程还没有Session,则创建一个新的Session
        if (s != null && s.getCustom()) {
            // 将获得的Session变量存储在ThreadLocal变量session里
            return s.getSession();
        }
        return null;
    }

    //关闭Session
    static void closeSession() throws HibernateException {
        SessionFactoryUtils.closeSession(getTlCurrentSession());
        sessions.set(null);
    }
}
//工具类 用于get和set session
public class HibernateUtil {

    //创建Session
    public static Session currentSession() throws HibernateException {
        return HwCurrentSessionContext.getCurrentSession();
    }

    //创建Session
    public static void setCurrentSession(Session session) throws HibernateException {
        HwCurrentSessionContext.setTlCurrentSession(session);
    }

    //关闭Session
    public static void closeCurrentSession() throws HibernateException {
        HwCurrentSessionContext.closeSession();
    }
}

2.2.3 配置hibernate.cfg.xml,用自定义的SessionContext替换SpringSessionContext

此时，系统会使用HwCurrentSessionContext作用hibernate的SessionContext,来为每次sql使用合适的currentSession.

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE hibernate-configuration PUBLIC
        "-//Hibernate/Hibernate Configuration DTD 3.0//EN"
        "http://www.hibernate.org/dtd/hibernate-configuration-3.0.dtd">
<hibernate-configuration>
    <session-factory>

        <!--自定义的SessionContext-->
        <property name="hibernate.current_session_context_class">com.??.hibernate.context.HwCurrentSessionContext</property>
....其他配置
    </session-factory>
</hibernate-configuration>

3. DEMO测试

在线程池中维护10个线程，并发保存10个随机的Employee到数据库。

在保存操作之前，手动调用 HibernateUtil.setCurrentSession为子线程配置currentSession
加上main线程在内共11个种使用了同一个session
为保证线程安全，对该session上锁

public class TestSessionFacade {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private final static int THREAD_NUM = 10;

    @Autowired
    private HibernateTemplate hibernateTemplate;
    @Autowired
    private SessionFactory    sessionFactory;

    public Response doAllInSession() {
        Session session = sessionFactory.getCurrentSession();
        hibernateTemplate.get(TestEmployee.class, 1);
        hibernateTemplate.save(EmployeeUtil.getRanE());
              // todo 经断点调试，以上get save 和以下10个线程中使用的session系同一个内存地址
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUM);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < THREAD_NUM; i++) {
            // todo: 开启10个线程，保存Employee
            executorService.submit(() -> {
                HibernateUtil.setCurrentSession(session); //为当前线程赋值session
                ThreadUtilKt.sleep(1000); //模拟每种资源业务的处理需要1秒
                synchronized (session) {
                    hibernateTemplate.save(EmployeeUtil.getRanE());
                }
            });
        }
        try {
            executorService.shutdown();
            executorService.awaitTermination(3, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (RandomUtil.getNum(4) % 2 == 1) {
            logger.error("========奇数回滚操作");
            session.doWork(Connection :: rollback);
        } else {
            logger.error("========偶数提交操作");
            session.doWork(Connection :: commit);
        }
        SoutUtil.print("Session还open的吗？ - " + session.isOpen());
        return Response.success("经debug调试确认使用了同一个session").add("COST_TIME", System.currentTimeMillis() - startTime);
    }

}

//respone输出如下
{"code":200,"success":true,"COST_TIME":1131,"message":"经debug调试确认使用了同一个session"}

由上代码和打印输出可以看出，10个并发线程，原本需要10秒以上的时间，只花了1.131秒就完成了所有10个employ的save操作

4. 总结

该demo已经实现了多个线程共享一个事务，且不干扰系统的正常运作。但由于不明确是否存在其他未知安全隐患，线上并没有做此优化。期待有大佬提供更权威的资料，或解决方案。

多线程案例 | 多线程共享同一个事务， 并发保存

1. 背景

2.DEMO实现

2.1 Employee实体

2.2 自定义hibernate的SessionContext --> 继承自SpringSessionContext

2.2.1 SessionContext的currentSession

2.2.2 自定义的SpringSessionContext --> HwCurrentSessionContext

2.2.3 配置hibernate.cfg.xml,用自定义的SessionContext替换SpringSessionContext

3. DEMO测试

4. 总结

推荐阅读更多精彩内容

多线程案例 | 多线程共享同一个事务，并发保存