多线程JPArepository批量插入

问题描述

我已经研究了一段时间的过程。在性能受到打击之前，过程运行良好。我想出了一种使其快速运行的方法，但是我真的不确定幕后发生了什么。现在它正在引发警告和错误，我不确定该怎么办。文件处理异常，但我不确定所有线程是否都已完成，并且我不认为我正确关闭了该应用程序。这就是您需要了解的一切...

使用缓冲读取器读取文件，然后对每条记录进行一些数据质量检查，对每条记录进行读取并通过数据质量检查，我们在文件中创建一个Java对象并将其插入List中。一旦List达到1000个对象大，我们将调用一个OracleService类，该类具有自动连接的Repo，并使用List执行saveAll方法。然后，我们继续读取文件并执行此操作，直到读取完文件为止。我将传递给服务和ExecutorService对象。因此，每次我们调用该服务时，都会得到一个包含我的对象的新List对象（该对象基本上是我们正在加载的表）和一个新的ExecutorService对象。进程运行正常，但是一旦我尝试关闭，就会抛出大量异常。这是我的所有代码...

我的Controller类的运行方法。这将从另一个实现CommandLineRunner的类中调用

    public void run() throws ParseException,IOException,InterruptedException {
        logger.info("******************** Aegis Check Inclearing DDA Trial Balance Table Load starting ********************");
        try (BufferedReader reader = new BufferedReader(new FileReader(inputFile))) {
            String line = reader.readLine();
            int count = 0;
            TrialBalanceBuilder builder = new TrialBalanceBuilder();
            while (line != null) {
                if (line.startsWith("D")) {
                    if (dataQuality(line)) {
                        TrialBalance trialBalance = builder.buildTrialBalanceObject(line,procDt,time);
                        insertList.add(trialBalance);
                        count++;
                        if (count == 1000) {
                            oracleService.loadToTableTrialBalance(insertList,executorService);
                            count = 0;
                            insertList.clear();
                        }
                    } else {
                        logger.info("Data quality check FAILED for record: " + line);
                        oracleService.revertInserts("DDA_TRIAL_BAL_STG",procDt.toString());
                        System.exit(111);
                    }
                }
                line = reader.readLine();
            }
            logger.info("Leftover record count is " + insertList.size());
            oracleService.loadToTableTrialBalance(insertList,executorService);
        } catch (IOException e) {
            e.printStackTrace();
        }

        logger.info("Updating Metadata table with new batch proc date");
        InclearingBatchMetadataBuilder inclearingBatchMetadataBuilder = new InclearingBatchMetadataBuilder();
        InclearingBatchMetadata inclearingBatchMetadata = inclearingBatchMetadataBuilder.buildInclearingBatchMetadataObject("DDA_TRIAL_BAL_STG",time,Constants.bankID);
        oracleService.insertBatchProcDtIntoMetaTable(inclearingBatchMetadata);
        logger.info("Successfully updated Metadata table with new batch proc date: " + procDt);

        Thread.sleep(10000);

        oracleService.cleanUpGOS("DDA_TRIAL_BAL_STG",1);

        executorService.shutdownNow();

        logger.info("******************** Aegis Check Inclearing DDA Trial Balance Table Load ended successfully ********************");
    }

我正在将ExecutorService对象传递给服务类。定义为...

    private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("Orders-%d").setDaemon(true).build();
    private ExecutorService executorService = Executors.newFixedThreadPool(10,threadFactory);

我的服务类别看起来就是这样。

@Service("oracleService")
public class OracleService {
    private static final Logger logger = LoggerFactory.getLogger(OracleService.class);

    @Autowired
    TrialBalanceRepo trialBalanceRepo;

    @Transactional
    public void loadToTableTrialBalance(List<TrialBalance> trialBalanceList,ExecutorService executorService) {
        logger.debug("Let's load to the database");
        logger.debug(trialBalanceList.toString());
        List<TrialBalance> multiThreadList = new ArrayList<>(trialBalanceList);
        try {
            executorService.execute(() -> trialBalanceRepo.saveAll(multiThreadList));
        } catch (ConcurrentModificationException | DataIntegrityViolationException ignored) {}
        logger.debug("Successfully loaded to database");
    }

然后在我的run方法中，调用该Service类中的其他方法，这些方法创建本地查询并在数据库上执行（用于清除等）

无论如何，我不知道线程何时完成。我发现在预生产阶段，当运行大量数据时，我们关闭了应用程序，并且并非所有数据都已完全加载。我也不知道这是否是最好的设计。我是否继续传递这些executorservice对象？这样做的全部目的是使最佳并行性进行下去，以便我们的性能更好。也许有更好的方法（最好不要重新设计整个应用程序并使用JPA之外的其他方法）

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）