天天看点
返回

springboot创建及使用多线程的几种方式

在数据处理中，多线程用到的场景很多，在满足计算机CPU处理能力的情况下，使用多线程可以明显提高程序运行效率，缩短大数据处理的能力。作为java程序开发，离不开spring,那么在spring中怎么创建多线程并将注册到spring的类在多线程中使用呢？我自己总结了一下，可以有两种方式，使用线程池和spring自带多线程注解使用。

使用线程池

我一般使用固定线程数量的线程池，假如数据量很大，我会将数据放到一个大集合中，然后按照一定的比例分配数目，同时我自己写了一个分页类，线程的数量可以根据分页类来自动调整。看代码:

分页类

package com.yiche.tag.model;

import java.util.List;

public class DataPage {
    private int page = 1; // 当前页

    public int totalPages = 0; // 总页数

    private int pageRecorders;// 每页5条数据

    private int totalRows = 0; // 总数据数

    private int pageStartRow = 0;// 每页的起始数

    private int pageEndRow = 0; // 每页显示数据的终止数

    private boolean hasNextPage = false; // 是否有下一页

    private boolean hasPreviousPage = false; // 是否有前一页

    private List list;

    // private Iterator it;

    public DataPage(List list, int pageRecorders) {
        init(list, pageRecorders);// 通过对象集，记录总数划分
    }

    /** *//**
     * 初始化list，并告之该list每页的记录数
     * @param list
     * @param pageRecorders
     */
    public void init(List list, int pageRecorders) {
        this.pageRecorders = pageRecorders;
        this.list = list;
        totalRows = list.size();
        // it = list.iterator();
        hasPreviousPage = false;
        if ((totalRows % pageRecorders) == 0) {
            totalPages = totalRows / pageRecorders;
        } else {
            totalPages = totalRows / pageRecorders + 1;
        }

        if (page >= totalPages) {
            hasNextPage = false;
        } else {
            hasNextPage = true;
        }

        if (totalRows < pageRecorders) {
            this.pageStartRow = 0;
            this.pageEndRow = totalRows;
        } else {
            this.pageStartRow = 0;
            this.pageEndRow = pageRecorders;
        }
    }


    // 判断要不要分页
    public boolean isNext() {
        return list.size() > 5;
    }

    public void setHasPreviousPage(boolean hasPreviousPage) {
        this.hasPreviousPage = hasPreviousPage;
    }

    public String toString(int temp) {
        String str = Integer.toString(temp);
        return str;
    }

    public void description() {

        String description = "共有数据数:" + this.getTotalRows() +

                "共有页数: " + this.getTotalPages() +

                "当前页数为:" + this.getPage() +

                " 是否有前一页: " + this.isHasPreviousPage() +

                " 是否有下一页:" + this.isHasNextPage() +

                " 开始行数:" + this.getPageStartRow() +

                " 终止行数:" + this.getPageEndRow();

        System.out.println(description);
    }

    public List getNextPage() {
        page = page + 1;

        disposePage();

        System.out.println("用户凋用的是第" + page + "页");
        this.description();
        return getObjects(page);
    }

    /** *//**
     * 处理分页
     */
    private void disposePage() {

        if (page == 0) {
            page = 1;
        }

        if ((page - 1) > 0) {
            hasPreviousPage = true;
        } else {
            hasPreviousPage = false;
        }

        if (page >= totalPages) {
            hasNextPage = false;
        } else {
            hasNextPage = true;
        }
    }

    public List getPreviousPage() {

        page = page - 1;

        if ((page - 1) > 0) {
            hasPreviousPage = true;
        } else {
            hasPreviousPage = false;
        }
        if (page >= totalPages) {
            hasNextPage = false;
        } else {
            hasNextPage = true;
        }
        this.description();
        return getObjects(page);
    }

    /** *//**
     * 获取第几页的内容
     *
     * @param page 从1开始
     * @return
     */
    public List getObjects(int page) {
        if (page == 0)
            this.setPage(1);
        else
            this.setPage(page);
        this.disposePage();
        if (page * pageRecorders < totalRows) {// 判断是否为最后一页
            pageEndRow = page * pageRecorders;
            pageStartRow = pageEndRow - pageRecorders;
        } else {
            pageEndRow = totalRows;
            pageStartRow = pageRecorders * (totalPages - 1);
        }

        List objects = null;
        if (!list.isEmpty()) {
            objects = list.subList(pageStartRow, pageEndRow);
        }
        //this.description();
        return objects;
    }

    public List getFistPage() {
        if (this.isNext()) {
            return list.subList(0, pageRecorders);
        } else {
            return list;
        }
    }

    public boolean isHasNextPage() {
        return hasNextPage;
    }


    public void setHasNextPage(boolean hasNextPage) {
        this.hasNextPage = hasNextPage;
    }


    public List getList() {
        return list;
    }


    public void setList(List list) {
        this.list = list;
    }


    public int getPage() {
        return page;
    }


    public void setPage(int page) {
        this.page = page;
    }


    public int getPageEndRow() {
        return pageEndRow;
    }


    public void setPageEndRow(int pageEndRow) {
        this.pageEndRow = pageEndRow;
    }


    public int getPageRecorders() {
        return pageRecorders;
    }


    public void setPageRecorders(int pageRecorders) {
        this.pageRecorders = pageRecorders;
    }


    public int getPageStartRow() {
        return pageStartRow;
    }


    public void setPageStartRow(int pageStartRow) {
        this.pageStartRow = pageStartRow;
    }


    public int getTotalPages() {
        return totalPages;
    }


    public void setTotalPages(int totalPages) {
        this.totalPages = totalPages;
    }


    public int getTotalRows() {
        return totalRows;
    }


    public void setTotalRows(int totalRows) {
        this.totalRows = totalRows;
    }


    public boolean isHasPreviousPage() {
        return hasPreviousPage;
    }

}

复制

这个分页类可以多个项目复用，既可以实现分页，还可以用到线程池中给线程分配数量使用。以下代码为如何使用该分页类给线程池使用的。

List<Integer> carModelIds = carsTagService.getAllCarModelIds();
        DataPage dataPage = new DataPage(carModelIds, 300);
        ExecutorService executorService = Executors.newFixedThreadPool(dataPage.totalPages);
        for (int i = 0; i < dataPage.totalPages; i++) {
            final int index = i + 1;
            executorService.execute(new Runnable() {
                List<Integer> temp = dataPage.getObjects(index);
                @Override
                public void run() {
                    try {
                        if (temp != null && temp.size() > 0) {
                            temp=Collections.synchronizedList(temp);
                            int sum = 0;
                            for (Integer carId : temp) {
                                try {
                                    carsTagService.insertCarTagByCarId(carId);
                                    sum++;
                                    if (sum % 100 == 0) {
                                        System.out.println("线程" + index + "当前执行了" + sum + "条,当前进度是" + sum * 100/ temp.size() + "%");
                                    }
                                } catch (Exception ex) {
                                    ex.printStackTrace();
                                }
                            }
                            System.out.println("线程" + index + "执行CarTagInitTask任务成功:" + temp.size());
                        } else {
                            System.out.println("CarTagInitTask任务执行完成:未读取到数据！");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        executorService.shutdown();
        while (true) {
            if (executorService.isTerminated()) {
                System.out.println("所有的子线程都结束了！");
                break;
            }
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

复制

上述代码中carModelIds就是要处理的数据集合，数目很大，如果直接for循环将会消耗很多的时间，利用线程池可以解决这个问题。但是如果直接创建多线程，线程中使用的对象需要final修饰，这对于spring管理的类不适用。使用线程池可以解决这个问题。

使用springboot自带@Async注解创建异步线程

在springboot中，可以使用@Async注解来将一个方法设置为异步方法，调用该方法的时候，是新开一个线程去调用。代码如下：

@Component
public class Task {
    @Async
    public void doTaskOne() throws Exception {
       System.out.println("开始做任务一");
        long start = System.currentTimeMillis();
        Thread.sleep(random.nextInt(10000));
        long end = System.currentTimeMillis();
        System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
    }
    @Async
    public void doTaskTwo() throws Exception {
        System.out.println("开始做任务二");
        long start = System.currentTimeMillis();
        Thread.sleep(random.nextInt(10000));
        long end = System.currentTimeMillis();
        System.out.println("完成任务二，耗时：" + (end - start) + "毫秒");
    }
    @Async
    public void doTaskThree() throws Exception {
       System.out.println("开始做任务三");
        long start = System.currentTimeMillis();
        Thread.sleep(random.nextInt(10000));
        long end = System.currentTimeMillis();
        System.out.println("完成任务三，耗时：" + (end - start) + "毫秒");
    }
}

复制

为了让@Async注解能够生效，还需要在Spring Boot的主程序中配置@EnableAsync，如下所示：

@SpringBootApplication
@EnableAsync
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

复制

此时可以反复执行单元测试，您可能会遇到各种不同的结果，比如：

  • 没有任何任务相关的输出
  • 有部分任务相关的输出
  • 乱序的任务相关的输出

原因是目前doTaskOne、doTaskTwo、doTaskThree三个函数的时候已经是异步执行了。主程序在异步调用之后，主程序并不会理会这三个函数是否执行完成了，由于没有其他需要执行的内容，所以程序就自动结束了，导致了不完整或是没有输出任务相关内容的情况。

注： @Async所修饰的函数不要定义为static类型，这样异步调用不会生效

异步回调

为了让doTaskOne、doTaskTwo、doTaskThree能正常结束，假设我们需要统计一下三个任务并发执行共耗时多少，这就需要等到上述三个函数都完成调动之后记录时间，并计算结果。

那么我们如何判断上述三个异步调用是否已经执行完成呢？我们需要使用Future<T>来返回异步调用的结果，就像如下方式改造doTaskOne函数：

@Async
public Future<String> doTaskOne() throws Exception {
    System.out.println("开始做任务一");
    long start = System.currentTimeMillis();
    Thread.sleep(random.nextInt(10000));
    long end = System.currentTimeMillis();
    System.out.println("完成任务一，耗时：" + (end - start) + "毫秒");
    return new AsyncResult<>("任务一完成");
}

复制

按照如上方式改造一下其他两个异步函数之后，下面我们改造一下测试用例，让测试在等待完成三个异步调用之后来做一些其他事情

@Test
public void test() throws Exception {
    long start = System.currentTimeMillis();
    Future<String> task1 = task.doTaskOne();
    Future<String> task2 = task.doTaskTwo();
    Future<String> task3 = task.doTaskThree();
    while(true) {
        if(task1.isDone() && task2.isDone() && task3.isDone()) {
            // 三个任务都调用完成，退出循环等待
            break;
        }
        Thread.sleep(1000);
    }
    long end = System.currentTimeMillis();
    System.out.println("任务全部完成，总耗时：" + (end - start) + "毫秒");
}

复制

看看我们做了哪些改变：

在测试用例一开始记录开始时间

在调用三个异步函数的时候，返回Future<String>类型的结果对象

在调用完三个异步函数之后，开启一个循环，根据返回的Future<String>对象来判断三个异步函数是否都结束了。若都结束，就结束循环；若没有都结束，就等1秒后再判断。

跳出循环之后，根据结束时间 - 开始时间，计算出三个任务并发执行的总耗时。

执行一下上述的单元测试，可以看到如下结果：

开始做任务一

开始做任务二

开始做任务三

完成任务三，耗时：37毫秒

完成任务二，耗时：3661毫秒

完成任务一，耗时：7149毫秒

任务全部完成，总耗时：8025毫秒

需要注意，spring管理的异步线程数量有限，如果是web项目的话，线程数量由tomcat的线程池配置有关系，所以如果可以，最好自己配置线程配置类。 

/**
 * springboot里面创建异步线程配置类
 * @author kouyy
 */
@Configuration
@EnableAsync
public class ThreadAsyncConfigurer implements AsyncConfigurer {

    @Bean
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
        //设置核心线程数
        threadPool.setCorePoolSize(10);
        //设置最大线程数
        threadPool.setMaxPoolSize(100);
        //线程池所使用的缓冲队列
        threadPool.setQueueCapacity(10);
        //等待任务在关机时完成--表明等待所有线程执行完
        threadPool.setWaitForTasksToCompleteOnShutdown(true);
        // 等待时间 （默认为0，此时立即停止），并没等待xx秒后强制停止
        threadPool.setAwaitTerminationSeconds(60);
        //  线程名称前缀
        threadPool.setThreadNamePrefix("MyAsync-");
        // 初始化线程
        threadPool.initialize();
        return threadPool;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
}

复制

配置类创建之后，以后再使用@Async创建异步线程就可以按照自己配置来使用了。不用担心和spring线程池数量冲突了。

上一篇: ThreadLocal内存泄漏问题及如何解决
下一篇: 【Maven学习】maven项目构建