从Excel批量导入数据说到ForkJoin的原理

码农飞哥

发布于 2021-08-18 10:45:35

1.7K0

前言

前面我们介绍了EasyPOI，用其进行Excel导入导出，非常的方便，非常的简单。但是4.0.0 版本以及其之前的版本都是通过单线程的方式对Excel中的数据进行解析的。效率比较差。今天我将做一个测试，5000条数据，分别使用EasyPOI的方式和自己手写ForkJoin的方式（多任务）的方式来导入，来比较这两者的性能差异。

测试前准备

1. 首先创建一个测试项目

首先我们需要创建一个测试项目，我这里新建了一个SpringBoot项目。然后引入easypoi的依赖，本次引入的easyPOI的版本是4.0.0版本。

<!--easypoi-->
        <dependency>
            <groupId>cn.afterturn</groupId>
            <artifactId>easypoi-spring-boot-starter</artifactId>
            <version>4.0.0</version>
        </dependency>
        <!--easypoi-->

2. 分别用两种方式实现导入

2.1：使用EasyPOI的方式

 @Override
    public String batchUploadStudent_easyPOI(MultipartFile file) throws Exception {
        long startTime = System.currentTimeMillis();
        List<Student> studentList = ExcelImportUtil.importExcel(file.getInputStream(), Student.class, new ImportParams());
        log.info("********通过EasyPOI读取文件总耗时是={},读取到的数据总条数是={}", (System.currentTimeMillis() - startTime) + "毫秒", studentList.size());
        return null;
    }

使用EasyPOI实现导入非常的简单，只需要调用importExcel方法即可。在此不再赘述。

2.2：自己手写Fork-Join的方式

接下来，我们自己手写Fork-Join的方式来实现文件的解析。

解析单元格的方法，本demo是直接挨个读取每个单元格的，当然也可以通过注解的方式来实现。代码如下：

    private List<Student> getData(Sheet sheet, int start, int end) {
        List<Student> mapList = new ArrayList<>();
        for (int i = start; i <= end; i++) {
            Student student = null;
            try {
                Row row = sheet.getRow(i);
                student = new Student();
                student.setClassName(ExcelUtil.getKeyValue(row.getCell(0)));
                student.setStudentName(ExcelUtil.getKeyValue(row.getCell(1)));
                student.setStudentMobile(ExcelUtil.getKeyValue(row.getCell(2)));               student.setIdCard(ExcelUtil.getKeyValue(row.getCell(3)));
                student.setStudentNo(ExcelUtil.getKeyValue(row.getCell(4)));
                student.setIdCard(ExcelUtil.getKeyValue(row.getCell(5)));
            } catch (Exception e) {
                log.info("***************税号={},文件名={},数据解析出现异常={}", e);
                continue;
            }
            mapList.add(student);
        }
        return mapList;
    }
}

这个方法也是很简单，就是读取开始行到结束行之间的所有数据。每个单元格的读取，严格按照Excel的字段顺序来读。

定义RecursiveTask类。

class JoinTask extends RecursiveTask<List<Student>> {
        //开始解读的行
        private int start;
        //结束解读的行
        private int end;
        //分页
        private Sheet sheet;
        //总的行数
        private int total;

        public JoinTask(int start, int end, Sheet sheet) {
            this.start = start;
            this.end = end;
            this.sheet = sheet;
            this.total = sheet.getLastRowNum();
        }

        @Override
        protected List<Student> compute() {
            //数据异常
            if (start > end || total < end) {
                return new ArrayList<>(1);
            }
            //每200行一个解析
            if (end - start <= 200) {
                return getData(sheet, start, end).stream().filter(DistinctUtil.distinctByKey(Student::getStudentNo)).collect(Collectors.toList());
            } else {
                //二分法，将数据平均分成两块
                int mid = (start + end) / 2;
                //递归调用，左边是序号小的那一块
                JoinTask rightTask = new JoinTask(start, mid, sheet);
                //递归调用，右边是数据大的那一块
                JoinTask leftTask = new JoinTask(mid + 1, end, sheet);
                //写法一
                rightTask.fork();
                List<Student> leftList =  leftTask.compute();
                List<Student> rightList = rightTask.join();
                //写法二
                //invokeAll(rightTask, leftTask);
                //List<Student> leftList = leftTask.join();
                //List<Student> rightList = rightTask.join();
                //将左边和右边的数据合并
                leftList.addAll(rightList);
                return leftList;
            }
        }
    }

RecursiveTask类是ForkJoin方式的核心类，后面会介绍这个类的作用。

调用的入口

    public List<Student> importExcel(Workbook workbook) {
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);;
        Sheet sheet = workbook.getSheetAt(0);
        //开启任务
        JoinTask joinTask = new JoinTask(1, sheet.getLastRowNum(), sheet);
        List<Student> importVOList = forkJoinPool.invoke(joinTask);
        //excel内部去重
        List<Student> noRepeatImportVOList = importVOList.stream().filter(DistinctUtil.distinctByKey(Student::getStudentNo)).collect(Collectors.toList());
        return noRepeatImportVOList;
    }

测试类：

  @Override
    public String batchUploadStudent_forkjoin(Workbook workbook) {
        long startTime = System.currentTimeMillis();
        List<Student> studentList = studentExcelImportWrapper.importExcel(workbook);
        log.info("********通过Fork-Join的方式读取文件总耗时是={},读取到的数据条数是={}", (System.currentTimeMillis() - startTime) + "毫秒", studentList.size());
        return null;
    }

3. 测试结果

上传同样的一个5000条数据的Excel,上传后的测试结果如下：

在这里插入图片描述从上测试结果，我们可以明显看出，性能差别还是挺大的，这主要是由于EasyPOI使用的是单线程的方式来读取Excel。数据量越大性能差别越明显。既然这个ForkJoin这么好用，那么就让我们来认识一下它吧。

ForkJoin初识

什么是ForkJoin框架

ForkJoin框架是Java7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。其中Fork就是将大任务拆分成若干个可以并发执行的小任务。Join就是合并所有小任务的执行结果。其执行流程如下图所示：

在这里插入图片描述

任务分割和结果合并说明

ForkJoinTask 就是最基本的任务，使用ForkJoin 框架必须创建的对象，它提供fork，join操作。一般而言，我们不需要直接继承它，只需要继承它的子类。它有两个子类。 RecursiveAction：用于无结果返回的任务。 RecursiveTask：用于有返回结果的任务。它的fork方法就是让task异步执行，join，就是让task同步执行，并获取返回值。

异常处理

ForkJoinTask在执行的时候可能会抛出异常，但是我们没有办法在主线程中直接捕获异常，所以ForkJoinTask提供了isCompleteAbnormally()方法来检查任务是否已经跑出异常或者已经被取消了。我们可以通过getException()方法获取异常信息，这个返回返回Throwable，如果任务被取消则返回CancellationException，如果任务正常执行完或者没有抛出异常，就返回null。

 if (rightTask.isCompletedAbnormally()) {
                    System.out.println(rightTask.getException());
                }

ForkJoin框架的实现原理

ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责存放程序提交到ForkJoinPool的任务，而ForkJoinWorkerThread数组负责执行这些任务。

fork方法

    public final ForkJoinTask<V> fork() {
        Thread t;
        //如果当前线程是ForkJoinWorkerThread线程
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
            ((ForkJoinWorkerThread)t).workQueue.push(this);
        else
            ForkJoinPool.common.externalPush(this);
        return this;
    }

如上代码：调用fork方法，这个方法的逻辑是，如果当前线程是ForkJoinWorkerThread线程则调用ForkJoinWorkerThread的push方法异步执行这个任务，然后立即返回结果。如果不是的话，则调用ForkJoinPool.common.externalPush(this);异步执行这个任务。

push方法

push方法是把当前任务存放在ForkJoinTask数组的queue里，然后再调用ForkJoinPool的signalWork()方法来唤醒或者创建一个工作线程来执行任务。

  final void push(ForkJoinTask<?> task) {
            ForkJoinTask<?>[] a; ForkJoinPool p;
            int b = base, s = top, n;
            if ((a = array) != null) {    // ignore if queue removed
                int m = a.length - 1;     // fenced write for task visibility
                //将当前任务存放在ForkJoinTask数组中
                U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
                U.putOrderedInt(this, QTOP, s + 1);
                if ((n = s - b) <= 1) {
                    if ((p = pool) != null)
                        //调用signalWork方法来唤醒或者创建一个工作线程来执行任务
                        p.signalWork(p.workQueues, this);
                }
                else if (n >= m)
                    growArray();
            }
        }

说完了fork()方法，接下来，让我们来看看join方法。

Join方法

ForkJoinTask的join方法的主要作用就是阻塞当前线程并等待执行结果

    public final V join() {
        int s;
        if ((s = doJoin() & DONE_MASK) != NORMAL)
            reportException(s);
        return getRawResult();
    }

        private void reportException(int s) {
        if (s == CANCELLED)
            throw new CancellationException();
        if (s == EXCEPTIONAL)
            rethrow(getThrowableException());
    }

如上，join()方法首先会调用doJoin()方法获取当前任务的执行状态来判断返回什么结果，任务的状态有四种：已完成（NORMAL），被取消（CANCELLED）,信号（SIGNAL）和出现异常（EXCEPTIONAL）。

如果任务状态不是已完成，则调用reportException方法，这个方法的逻辑是如果任务是已取消，则抛出CancellationException异常如果任务出现异常，则抛出封装重抛异常。
如果任务是已完成，则返回结果。接下来，让我们来看看doJoin()方法。

doJoin方法

    private int doJoin() {
        int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
        return (s = status) < 0 ? s :
            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
            (w = (wt = (ForkJoinWorkerThread)t).workQueue).
            tryUnpush(this) && (s = doExec()) < 0 ? s :
            wt.pool.awaitJoin(w, this, 0L) :
            externalAwaitDone();
    }

在doJoin()方法里，首先通过查看任务的状态，看任务是否执行完了，如果执行完了，则直接返回任务状态，如果没有执行完，则从任务数组里取出任务并执行。如果任务顺利执行完成了，则设置任务的状态为NORMAL，如果出现异常，则记录异常，并将任务状态设置为EXCEPTIONAL。

参考资料

JDK1.8 源码 http://ifeve.com/fork-join-5/

源码地址

https://github.com/XWxiaowei/JavaCode/tree/master/spring-poi-demo

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-04-18，如有侵权请联系 cloudcommunity@tencent.com 删除

unix

编程算法

腾讯云测试服务