事务与多线程的坑及调优Tips

原由及介绍

在处理原始对帐文件的时候，我将数据归类后批量存入相应的表中。在持久化的时候，用了parallelStream()，想着同时存入不少表这样能够提升效率。

@Override
@Transactional
public boolean handleTask(AccEbankAlEveBill[] task, String ownSign) throws Exception {
    Arrays.stream(task).parallel().map(AccEbankBill::convert).collect(Collectors.groupingBy(AccEbankBill::getTradeType)).entrySet()
    // 这里出了问题
    .parallelStream().forEach(item -> {
        switch (EbankAleveTradeTypeEnum.valueOf(item.getKey())) {
        // 提现...
        // 充值
        case RECHARGE:
            accEbankRechargeBillDao.batchSave(item.getValue());
            break;

        // 利息
        case INTEREST:
            accEbankInterestBillDao.batchSave(item.getValue());
            break;
        case OTHERS:
            accEBankAleveOthersBillDao.batchSave(item.getValue());
            break;
        }
    });

    // DB update
    return true;
}

问题分析

上面代码中将对帐数据按类型归类后，获得一个Map<String, List<Bean>>，key为类型，value为数据。考虑到数据类型有不少种，而后使用了parallelStream()。在最近的一次自测中，因为开发数据库网络问题，形成事务处理超时，但发现数据没有回滚！

开始沿着数据库超时中断机制的思路找问题，花了比较多的时间在分析是数据库端触发了中断，仍是应用层主动中断，以及二者对是否回滚有啥区别。。最后发现这些都不是缘由

次日一早忽然想到这里的parallelStream()多是罪魁祸首，由于它开启了多线程（多线程每每有问题），本机环境一共有4个worker在处理（包含主线程），可是超时致使的org.springframework.transaction.TransactionTimedOutException错误是发生在主线程内的，那确定只有主线程回滚了。后经测试证明。

总结

解决方法就是去除parallelStream()，简简单单的使用Map的forEach就行了。

结论：事务只能管着开启事务的线程，其余子线程出了问题都感知不到，因此在多线程环境操做DB要慎重。普通的多线程很容易发觉，但parallelStream是也是，切记

调优Tips

1. 线程池的大小

线程池的大小 = 处理器的核的数目 指望的CPU利用率 (1 + W／C)

其中：

• CPU利用率介于0和1之间
• W/C是等待时间与计算时间的比率

源自《Java并发编程实战》Brian Goetz的建议

2. 文件下载

HTTP(S)用apache httpclient可实现连接池和断点续传， FTP可以使用Apache Commons Net API。重试间隔设置为5～10分钟较合适。高频容易搞死服务器，低频会阻塞自身程序。重试次数和超时时间根据业务状况设置。