多线程快速导入大量数据到mysql

加入事物都成功则成功,其中一个环节失败则回滚

    @Autowired
    private UserTestService userTestService;

    @Autowired
    private UserTestItemService userTestItemService;

    @Autowired
    private PlatformTransactionManager platformTransactionManager;
    @Autowired
    private TransactionDefinition transactionDefinition;


    
    //@Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)
    @RequestMapping(value = "/save1/{size1}")
    public String save(@PathVariable("size1") Integer size1) throws ExecutionException, InterruptedException {
        // 事务
        List statusList = new ArrayList<>();
        long startTime = System.currentTimeMillis();
        log.info("开始时间[{}]", startTime);
        List userTests = createDate(size1);
        //开启10个线程
        int nThreads = 10;
        //过大list的数据总条数
        final int size = userTests.size();
        //开启一个有10个线程的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
        //通过实现Callback接口，并用Future可以来接收多线程的执行结果
        List> futures = new ArrayList<>(nThreads);
        //遍历生成50个线程来执行任务，每个线程处理50条数据
        for (int i = 0; i < nThreads; i++) {
            final List allusersEntityImportList = userTests.subList(size / nThreads * i, size / nThreads * (i + 1));
            //线程执行，实现callable接口，批量保存
            CompletableFuture exceptionally = CompletableFuture.supplyAsync(() -> {
                // 事务传递 TransactionDefinition.PROPAGATION_REQUIRES_NEW = 3
                statusList.add(platformTransactionManager.getTransaction(new DefaultTransactionDefinition(3)));
                log.info("主表线程名称是=[{}],线程id=[{}]",Thread.currentThread().getName(),Thread.currentThread().getId());
                //保存主表信息
                //int a=10/0;
                userTestService.save(allusersEntityImportList);
                return allusersEntityImportList;
            }).thenApply(list -> {
                try {
                    int a=10/0;
                    log.info("从表线程名称是=[{}],线程id=[{}]",Thread.currentThread().getName(),Thread.currentThread().getId());
                    List dateItem = createDateItem(list);
                    int save = userTestItemService.save(dateItem);
                    return save;
                }catch (Exception e){
                    throw new NullPointerException(e.getMessage());
                }
            }).exceptionally(error -> {
                System.out.println("发生了异常信息");
                throw new NullPointerException(error.getMessage());
            });
            //每50条数据提交一次任务，报异常则执行事务回滚，并将结果添加到furures中用于展示
            futures.add(exceptionally);
        }

        try {
            // List列表转成CompletableFuture的Array数组,使其可以作为allOf()的参数
            // 使用join()方法使得主线程阻塞，并等待所有并行线程完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{})).join();
        }catch (Exception e){
            log.error("1111111");
            // 异常进行所有子任务回滚
            statusList.forEach(i->platformTransactionManager.rollback(i));
        }
        //执行提交
        statusList.forEach(i->platformTransactionManager.commit(i));
        executorService.shutdown();

        long endTime = System.currentTimeMillis();
        log.info("结束时间[{}]", startTime - endTime);

        return "2002";
    }

**单个CompletableFuture controll代码
注意事务注解Transactional要写在servicce 写在controller会失效
**

       @RequestMapping(value = "/save2/{size2}")
    public String save(@PathVariable("size2") Integer size2) {
        long startTime = System.currentTimeMillis();
        log.info("开始时间[{}]", startTime);
        List userTests = createDate(size2);
        //开启10个线程1
        int nThreads = 10;
        //过大list的数据总条数
        final int size = userTests.size();
        //开启一个有10个线程的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
        //通过实现Callback接口，并用Future可以来接收多线程的执行结果
        List> futures = new ArrayList<>(nThreads);
        //遍历生成50个线程来执行任务，每个线程处理50条数据
        for (int i = 0; i < nThreads; i++) {
            final List allusersEntityImportList = userTests.subList(size / nThreads * i, size / nThreads * (i + 1));
            //线程执行，实现callable接口，批量保存
            CompletableFuture exceptionally = CompletableFuture.supplyAsync(() -> {
                log.info("主表线程名称是=[{}],线程id=[{}]", Thread.currentThread().getName(), Thread.currentThread().getId());
                //保存主表信息
                userTestService.save(allusersEntityImportList);
                return 1;
            }).exceptionally(error -> {
                System.out.println("发生了异常信息[]===");
                throw new NullPointerException(error.getMessage());
            });
            //每50条数据提交一次任务，报异常则执行事务回滚，并将结果添加到furures中用于展示
            futures.add(exceptionally);
        }
        // List列表转成CompletableFuture的Array数组,使其可以作为allOf()的参数
        // 使用join()方法使得主线程阻塞，并等待所有并行线程完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{})).join();

        executorService.shutdown();

        long endTime = System.currentTimeMillis();
        log.info("结束时间[{}]", startTime - endTime);

        return "2002";
    }

service代码

    @Transactional(rollbackFor = Exception.class)
    @Override
    public List save(List userTests) {
        userTestMapper.insertList(userTests);
        List dateItem = createDateItem(userTests);
        if (Thread.currentThread().getId()==90){
            int a=10/0;
        }
        userTestItemService.save(dateItem);
        return null;
    }