亲宝软件园·资讯

展开

Java线程池并发执行多任务

micro_hz 人气:0

Java线程池并发执行多个任务

Java在语言层面提供了多线程的支持,线程池能够避免频繁的线程创建和销毁的开销,因此很多时候在项目当中我们是使用的线程池去完成多线程的任务。

Java提供了Executors 框架提供了一些基础的组件能够轻松的完成多线程异步的操作,Executors提供了一系列的静态工厂方法能够获取不同的ExecutorService实现,ExecutorService扩展了Executors接口,Executors相当简单:

public interface Executor {
    void execute(Runnable command);
}

把任务本身和任务的执行解耦了,如果说Runnable是可异步执行任务的抽象,那Executor就是如何执行可异步执行任务的抽象,说起来比较绕口。

本文不讲解线程的一些基础知识,因为网上的其他文章已经写的足够详细和泛滥。我写写多个异步任务的并发执行与结果的获取问题。

假设这样一个场景:我们要组装一个对象,这个对象由大量小的内容组成,这些内容是无关联无依赖关系的,如果我们串行的去执行,如果每个任务耗时10秒钟,一共有10个任务,那我们就需要100秒才能获取到结果。显然我们可以采用线程池,每个任务起一个线程,忽略线程启动时间,我们只需要10秒钟就能获取到结果。这里还有两种选择,这10秒钟我们可以去做其他事,也可以等待结果。

我们来完成这样的操作:

// 这是任务的抽象
class GetContentTask implements Callable<String> {
        
        private String name;
        
        private Integer sleepTimes;
        
        public GetContentTask(String name, Integer sleepTimes) {
            this.name = name;
            this.sleepTimes = sleepTimes;
        }
        public String call() throws Exception {
            // 假设这是一个比较耗时的操作
            Thread.sleep(sleepTimes * 1000);
            return "this is content : hello " + this.name;
        }
        
    }

采用completionService :

// 方法一
        ExecutorService executorService = Executors.newCachedThreadPool();
        CompletionService<String> completionService = new ExecutorCompletionService(executorService);
        ExecuteServiceDemo executeServiceDemo = new ExecuteServiceDemo();
        // 十个
        long startTime = System.currentTimeMillis();
        int count = 0;
        for (int i = 0;i < 10;i ++) {
            count ++;
            GetContentTask getContentTask = new ExecuteServiceDemo.GetContentTask("micro" + i, 10);
            completionService.submit(getContentTask);
        }
        System.out.println("提交完任务,主线程空闲了, 可以去做一些事情。");
        // 假装做了8秒种其他事情
        try {
            Thread.sleep(8000);
            System.out.println("主线程做完了,等待结果");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            // 做完事情要结果
            for (int i = 0;i < count;i ++) {
                Future<String> result = completionService.take();
                System.out.println(result.get());
            }
            long endTime = System.currentTimeMillis();
            System.out.println("耗时 : " + (endTime - startTime) / 1000);
        }  catch (Exception ex) {
            System.out.println(ex.getMessage());
        }

执行结果为:

提交完任务,主线程空闲了, 可以去做一些事情。
主线程做完了,等待结果
this is content : hello micro9
this is content : hello micro7
this is content : hello micro2
this is content : hello micro5
this is content : hello micro4
this is content : hello micro8
this is content : hello micro1
this is content : hello micro3
this is content : hello micro0
this is content : hello micro6
耗时 : 10

如果多个不想一个一个提交,可以采用 invokeAll一并提交,但是会同步等待这些任务

// 方法二
        ExecutorService executeService = Executors.newCachedThreadPool();
        List<GetContentTask> taskList = new ArrayList<GetContentTask>();
        long startTime = System.currentTimeMillis();
        for (int i = 0;i < 10;i ++) {
            taskList.add(new GetContentTask("micro" + i, 10));
        }
        try {
            System.out.println("主线程发起异步任务请求");
            List<Future<String>> resultList = executeService.invokeAll(taskList);
            // 这里会阻塞等待resultList获取到所有异步执行的结果才会执行 
            for (Future<String> future : resultList) {
                System.out.println(future.get());
            }
            // 主线程假装很忙执行8秒钟
            Thread.sleep(8);
            long endTime = System.currentTimeMillis();
            System.out.println("耗时 : " + (endTime - startTime) / 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }

主线程发起异步任务请求
this is content : hello micro0
this is content : hello micro1
this is content : hello micro2
this is content : hello micro3
this is content : hello micro4
this is content : hello micro5
this is content : hello micro6
this is content : hello micro7
this is content : hello micro8
this is content : hello micro9
耗时 : 10

如果一系列请求,我们并不需要等待每个请求,我们可以invokeAny,只要某一个请求返回即可。

ExecutorService executorService = Executors.newCachedThreadPool();
        ArrayList<GetContentTask> taskList = new ArrayList<GetContentTask>();
        taskList.add(new GetContentTask("micro1",3));
        taskList.add(new GetContentTask("micro2", 6));
        try {
            List<Future<String>> resultList = executorService.invokeAll(taskList);// 等待6秒 
//            String result2 = executorService.invokeAny(taskList); // 等待3秒
            // invokeAll 提交一堆任务并行处理并拿到结果
            // invokeAny就是提交一堆并行任务拿到一个结果即可
            for (Future<String> result : resultList) {
                System.out.println(result.get());
            }
//            System.out.println(result2);
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("主线程等待");

如果我虽然发送了一堆异步的任务,但是我只等待一定的时间,在规定的时间没有返回我就不要了,例如很多时候的网络请求其他服务器如果要数据,由于网络原因不能一直等待,在规定时间内去拿,拿不到就我使用一个默认值。

这样的场景,我们可以使用下面的写法:

try {
            ExecutorService executorService = Executors.newCachedThreadPool();
            List<Callable<String>> taskList = new ArrayList<Callable<String>>();
            taskList.add(new GetContentTask("micro1", 4));
            taskList.add(new GetContentTask("micro2", 6));
            // 等待五秒
            List<Future<String>> resultList = executorService.invokeAll(taskList, 5, TimeUnit.SECONDS);
            for (Future<String> future : resultList) {
                System.out.println(future.get());
            }
        } catch (Exception e) {
            e.printStackTrace();
        } 

this is content : hello micro1
java.util.concurrent.CancellationException
    at java.util.concurrent.FutureTask.report(FutureTask.java:121)
    at java.util.concurrent.FutureTask.get(FutureTask.java:192)
    at com.micro.demo.spring.ExecuteServiceDemo.main(ExecuteServiceDemo.java:105)

因为只等待5秒,6秒的那个任务自然获取不到,抛出异常,如果将等待时间设置成8秒,就都能获取到。

Java线程池的正确使用

线程可认为是操作系统可调度的最小的程序执行序列,一般作为进程的组成部分,同一进程中多个线程可共享该进程的资源(如内存等)。JVM线程跟内核轻量级进程有一对一的映射关系,所以JVM中的线程是很宝贵的。

一般在工程上多线程的实现是基于线程池的。因为相比自己创建线程,多线程具有以下优点:

1.Executors存在什么问题

看阿里巴巴开发手册并发编程这块有一条:线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式。

2.Executors为什么存在缺陷

2.1 线程池工作原理

当一个任务通过execute(Runnable)方法欲添加到线程池时:

2.2 newFixedThreadPool分析

Java中的BlockingQueue主要有两种实现,分别是ArrayBlockingQueue 和 LinkedBlockingQueue。

ArrayBlockingQueue是一个用数组实现的有界阻塞队列,必须设置容量。

LinkedBlockingQueue是一个用链表实现的有界阻塞队列,容量可以选择进行设置,不设置的话,将是一个无边界的阻塞队列,最大长度为Integer.MAX_VALUE。

这里的问题就出在:不设置的话,将是一个无边界的阻塞队列,最大长度为Integer.MAX_VALUE。也就是说,如果我们不设置LinkedBlockingQueue的容量的话,其默认容量将会是Integer.MAX_VALUE。而newFixedThreadPool中创建LinkedBlockingQueue时,并未指定容量。此时,LinkedBlockingQueue就是一个无边界队列,对于一个无边界队列来说,是可以不断的向队列中加入任务的,这种情况下就有可能因为任务过多而导致内存溢出问题。

2.3 newCachedThreadPool分析

结合上述流程图,核心线程数=0,最大线程无限大,由于SynchronousQueue是一个缓存值为1的阻塞队列。当有大量任务请求时,线程池会创建大量线程,造成OOM。

3. 线程池参数详解

3.1 构造方法

3.2 线程池拒绝策略

RejectedExecutionHandler(饱和策略):当队列和线程池都满了,说明线程池处于饱和状态,那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy,表示无法处理新任务时抛出异常。。以下是JDK1.5提供的四种策略。

4. 线程池正确打开方式

4.1 创建线程池

避免使用Executors创建线程池,主要是避免使用其中的默认实现,那么我们可以自己直接调用ThreadPoolExecutor的构造函数来自己创建线程池。在创建的同时,给BlockQueue指定容量就可以了。

4.2 向线程池提交任务

我们可以使用execute提交的任务,但是execute方法没有返回值,所以无法判断任务知否被线程池执行成功。通过以下代码可知execute方法输入的任务是一个Runnable类的实例。

我们也可以使用submit 方法来提交任务,它会返回一个future,那么我们可以通过这个future来判断任务是否执行成功,通过future的get方法来获取返回值,get方法会阻塞住直到任务完成,而使用get(long timeout, TimeUnit unit)方法则会阻塞一段时间后立即返回,这时有可能任务没有执行完。

4.3 关闭线程池

shutdown关闭线程池

方法定义:public void shutdown()

注意:这个函数不会等待提交的任务执行完成,要想等待全部任务完成,可以调用:

public boolean awaitTermination(longtimeout, TimeUnit unit)

shutdownNow关闭线程池并中断任务

方法定义:public List shutdownNow()

4.4 如何配置线程池大小

CPU密集型任务

IO密集型任务

以上为个人经验,希望能给大家一个参考,也希望大家多多支持。

加载全部内容

相关教程
猜你喜欢
用户评论