Java多线程系列七——ExecutorService

java.util.concurrent.ExecutorService接口提供了许多线程管理的方法

Method	说明
shutdown	拒绝接收新的任务，待已提交的任务执行后关闭，且宿主线程不阻塞，若需要阻塞可借助awaitTermination实现
shutdownNow	停止所有正在执行的任务，挂起未执行的任务并关闭，且宿主线程不阻塞，若需要阻塞可借助awaitTermination实现
awaitTermination	当发生shutdown时，阻塞宿主线程直到约定的时间已过或者所有任务完成
submit	提交任务Callable/Runnable，可利用Future的get()方法使宿主线程阻塞直到任务结束后返回结果

有了以上方法，便可以基于此接口实现线程池的各种功能(例如java.util.concurrent.ThreadPoolExecutor/java.util.concurrent.ScheduledThreadPoolExecutor)，以java.util.concurrent.ThreadPoolExecutor为例，其参数的详解

Name	Type	说明
corePoolSize	int	线程池中最小的线程数
maximumPoolSize	int	线程池中最大的线程数
keepAliveTime	long	线程空闲时间，若线程数大于corePoolSize，空闲时间超过该值的线程将被终止回收
unit	TimeUnit	keepAliveTime的时间单位
workQueue	BlockingQueue<Runnable>	已提交但未执行的任务队列
threadFactory	ThreadFactory	创建新线程的工厂
handler	RejectedExecutionHandler	当线程池或队列达到上限拒绝新任务抛出异常时的处理类

同时，java.util.concurrent.Executors类提供的常用方法有

Method	说明	基类
newFixedThreadPool	线程池中含固定数量的线程	基于java.util.concurrent.ThreadPoolExecutor类
newSingleThreadExecutor	线程池中仅含一个工作线程
newCachedThreadPool	按需创建线程，若线程池中无可用线程，则创建新的线程并加入，直到线程数达到上限值(Integer.MAX_VALUE)
newWorkStealingPool	按照可用CPU数创建线程池	基于java.util.concurrent.ForkJoinPool类

java.util.concurrent.ForkJoinPool类是Fork/Join框架的实现类，Fork/Join框架是Java7提供了的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架，该类在有递归实现的场景有更优异的表现。

测试代码如下

import java.util.Date;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import org.junit.Assert;

import org.junit.Test;

/**

 * @Description: 测试ExecutorService

 */

public class ThreadExecutorServiceTest {

    private static final String THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION = "This is shutdownWithAwaitTermination";

    private static final int RESULT = 111;

    private static boolean submitRunnable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Future<?> future = executorService.submit(new Runnable() {

            @Override

            public void run() {

                System.out.println("This is submitRunnable");

            }

        });

        return future.get() == null;

    }

    private static Integer submitRunnableWithResult() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Future<Integer> future = executorService.submit(new Runnable() {

            @Override

            public void run() {

                System.out.println("This is submitRunnableWithResult");

            }

        }, RESULT);

        return future.get();

    }

    private static Integer submitBlockCallable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        Future<Integer> future = executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                System.out.println("This is submitBlockCallable");

                return RESULT;

            }

        });

        return future.get();// 阻塞

    }

    private static boolean submitNonBlockCallable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        Future<Integer> future = executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                System.out.println("This is submitNonBlockCallable");

                return RESULT;

            }

        });

        while (!future.isDone()) {// 非阻塞

            System.out.println(new Date());

        }

        return future.isDone();

    }

    private static String shutdown() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        final StringBuilder sb = new StringBuilder();

        executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                Thread.sleep(10000);

                sb.append("This is shutdown");

                return RESULT;

            }

        });

        executorService.shutdown();

        return sb.toString();

    }

    private static String shutdownWithAwaitTermination() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        final StringBuilder sb = new StringBuilder();

        executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                Thread.sleep(10000);

                sb.append(THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION);

                return RESULT;

            }

        });

        executorService.shutdown();

        executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS);

        return sb.toString();

    }

    @Test

    public void test() throws InterruptedException, ExecutionException {

        Assert.assertTrue(submitRunnable());

        Assert.assertEquals(RESULT, submitRunnableWithResult().intValue());

        Assert.assertEquals(RESULT, submitBlockCallable().intValue());

        Assert.assertTrue(submitNonBlockCallable());

        Assert.assertTrue(shutdown().isEmpty());

        Assert.assertEquals(THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION, shutdownWithAwaitTermination());

    }

}

---恢复内容结束---

java.util.concurrent.ExecutorService接口提供了许多线程管理的方法

Method	说明
shutdown	拒绝接收新的任务，待已提交的任务执行后关闭，且宿主线程不阻塞，若需要阻塞可借助awaitTermination实现
shutdownNow	停止所有正在执行的任务，挂起未执行的任务并关闭，且宿主线程不阻塞，若需要阻塞可借助awaitTermination实现
awaitTermination	当发生shutdown时，阻塞宿主线程直到约定的时间已过或者所有任务完成
submit	提交任务Callable/Runnable，可利用Future的get()方法使宿主线程阻塞直到任务结束后返回结果

Name	Type	说明
corePoolSize	int	线程池中最小的线程数
maximumPoolSize	int	线程池中最大的线程数
keepAliveTime	long	线程空闲时间，若线程数大于corePoolSize，空闲时间超过该值的线程将被终止回收
unit	TimeUnit	keepAliveTime的时间单位
workQueue	BlockingQueue<Runnable>	已提交但未执行的任务队列
threadFactory	ThreadFactory	创建新线程的工厂
handler	RejectedExecutionHandler	当线程池或队列达到上限拒绝新任务抛出异常时的处理类

同时，java.util.concurrent.Executors类提供了基于java.util.concurrent.ThreadPoolExecutor类的工具方法，常用方法有

Method	说明
newFixedThreadPool	线程池中含固定数量的线程
newSingleThreadExecutor	线程池中仅含一个工作线程
newCachedThreadPool	按需创建线程，若线程池中无可用线程，则创建新的线程并加入，直到线程数达到上限值(Integer.MAX_VALUE)

测试代码如下

import java.util.Arrays;

import java.util.Date;

import java.util.List;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import org.junit.Assert;

import org.junit.Test;

import org.lxp.multiple.thread.task.SumTask;

/**

 * @Description: 测试ExecutorService

 * @author Super.Li

 * @date Jul 6, 2017

 */

public class ThreadExecutorServiceTest {

    private static final String THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION = "This is shutdownWithAwaitTermination";

    private static final int RESULT = 111;

    private static boolean submitRunnable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Future<?> future = executorService.submit(new Runnable() {

            @Override

            public void run() {

                System.out.println("This is submitRunnable");

            }

        });

        return future.get() == null;

    }

    private static Integer submitRunnableWithResult() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Future<Integer> future = executorService.submit(new Runnable() {

            @Override

            public void run() {

                System.out.println("This is submitRunnableWithResult");

            }

        }, RESULT);

        return future.get();

    }

    private static Integer submitBlockCallable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        Future<Integer> future = executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                System.out.println("This is submitBlockCallable");

                return RESULT;

            }

        });

        return future.get();// 阻塞

    }

    private static boolean submitNonBlockCallable() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        Future<Integer> future = executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                System.out.println("This is submitNonBlockCallable");

                return RESULT;

            }

        });

        while (!future.isDone()) {// 非阻塞

            System.out.println(new Date());

        }

        return future.isDone();

    }

    private static String shutdown() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        final StringBuilder sb = new StringBuilder();

        executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                Thread.sleep(10000);

                sb.append("This is shutdown");

                return RESULT;

            }

        });

        executorService.shutdown();

        return sb.toString();

    }

    private static String shutdownWithAwaitTermination() throws InterruptedException, ExecutionException {

        ExecutorService executorService = Executors.newFixedThreadPool(1);

        final StringBuilder sb = new StringBuilder();

        executorService.submit(new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                Thread.sleep(10000);

                sb.append(THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION);

                return RESULT;

            }

        });

        executorService.shutdown();

        executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS);

        return sb.toString();

    }

    private static int testForkJoinPool(List<Integer> list) throws InterruptedException, ExecutionException {

        ForkJoinPool forkJoinPool = new ForkJoinPool(8);

        Future<Integer> future = forkJoinPool.submit(new SumTask(list));

        return future.get();

    }

    @Test

    public void test() throws InterruptedException, ExecutionException {

        Assert.assertTrue(submitRunnable());

        Assert.assertEquals(RESULT, submitRunnableWithResult().intValue());

        Assert.assertEquals(RESULT, submitBlockCallable().intValue());

        Assert.assertTrue(submitNonBlockCallable());

        Assert.assertTrue(shutdown().isEmpty());

        Assert.assertEquals(THIS_IS_SHUTDOWN_WITH_AWAIT_TERMINATION, shutdownWithAwaitTermination());

        Assert.assertEquals(10, testForkJoinPool(Arrays.asList(new Integer[] { 1, 2, 3, 4 })));

        Assert.assertEquals(49, testForkJoinPool(Arrays.asList(new Integer[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 })));

        Assert.assertEquals(60, testForkJoinPool(Arrays.asList(new Integer[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11 })));

    }

}

SumTask类如下：

import java.util.List;

import java.util.concurrent.RecursiveTask;

public class SumTask extends RecursiveTask<Integer> {

    private static final long serialVersionUID = 1L;

    private List<Integer> list;

    public SumTask(List<Integer> list) {

        this.list = list;

    }

    /**

     * Ensure it is necessary to divide the job to parts and finish them separately

     *

     * @return

     */

    @Override

    protected Integer compute() {

        int rtn, size = list.size();

        if (size < 10) {

            rtn = sum(list);

        } else {

            SumTask subTask1 = new SumTask(list.subList(0, size / 2));

            SumTask subTask2 = new SumTask(list.subList(size / 2 + 1, size));

            subTask1.fork();

            subTask2.fork();

            rtn = subTask1.join() + subTask2.join();

        }

        return rtn;

    }

    private int sum(List<Integer> list) {

        return list.stream().mapToInt(number -> number.intValue()).sum();

    }

}

Java多线程系列七——ExecutorService的更多相关文章

java多线程系列(七)---Callable、Future和FutureTask
Callable.Future和FutureTask 前言:如有不正确的地方,还望指正. 目录认识cpu.核心与线程 java多线程系列(一)之java多线程技能 java多线程系列(二)之对象变量 ...
【Java多线程系列七】ExecutorService
java.util.concurrent.ExecutorService接口提供了许多线程管理的方法 Method 说明 shutdown 拒绝接收新的任务,待已提交的任务执行后关闭,且宿主线程不阻塞 ...
（Java多线程系列七）Java内存模型和线程的三大特性
Java内存模型和线程的三大特性多线程有三大特性:原子性.可见性.有序性 1.Java内存模型 Java内存模型(Java Memory Model ,JMM),决定一个线程对共享变量的写入时,能对 ...
java多线程系列(八)---CountDownLatch和CyclicBarrie
CountDownLatch 前言:如有不正确的地方,还望指正. 目录认识cpu.核心与线程 java多线程系列(一)之java多线程技能 java多线程系列(二)之对象变量的并发访问 java多线 ...
java多线程系列(九)---ArrayBlockingQueue源码分析
java多线程系列(九)---ArrayBlockingQueue源码分析目录认识cpu.核心与线程 java多线程系列(一)之java多线程技能 java多线程系列(二)之对象变量的并发访问 j ...
Java多线程系列--&OpenCurlyDoubleQuote;JUC线程池”06之 Callable和Future
概要本章介绍线程池中的Callable和Future.Callable 和 Future 简介示例和源码分析(基于JDK1.7.0_40) 转载请注明出处:http://www.cnblogs.co ...
Java多线程系列--&OpenCurlyDoubleQuote;JUC锁”11之 Semaphore信号量的原理和示例
概要本章,我们对JUC包中的信号量Semaphore进行学习.内容包括:Semaphore简介Semaphore数据结构Semaphore源码分析(基于JDK1.7.0_40)Semaphore示例 ...
Java多线程系列--&OpenCurlyDoubleQuote;JUC线程池”01之线程池架构
概要前面分别介绍了"Java多线程基础"."JUC原子类"和"JUC锁".本章介绍JUC的最后一部分的内容——线程池.内容包括:线程池架构 ...
Java多线程系列--&OpenCurlyDoubleQuote;JUC线程池”03之线程池原理(二)
概要在前面一章"Java多线程系列--“JUC线程池”02之线程池原理(一)"中介绍了线程池的数据结构,本章会通过分析线程池的源码,对线程池进行说明.内容包括:线程池示例参考代 ...

随机推荐

ios中，在SearchBar里面搜索内容，可根据内容来查找所需的信息资源，可获得SearchBar中的内容
贴一段我很久以前写的小demo,你们就明白了,是把textField套在alertView里的@interface ViewController : UIViewController <UIAl ...
C语言宏定义时&num;(井号)和&num;&num;(双井号)的用法1
#在英语里面叫做 pound 在C语言的宏定义中,一个#表示字符串化:两个#代表concatenate 举例如下: #include <iostream> void quit_comman ...
netbeans环境的建立
这是一个简单的实验,熟悉NetBeans的IDE环境的开发首先下载一个NetBeans,可以在官网上下https://netbeans.org/downloads/index.html 要装NetB ...
PS 颜色表大全-CMYK颜色表（2）
CMYK颜色表编号 C M Y K R G B 16进制值 1 0 100 100 45 139 0 22 8B0016 2 0 100 100 25 178 0 31 B2001F 3 0 100 ...
Android之如何混淆代码和相关配置
昨天,客户想看一下目前项目开发到什么程度了,于是需要将项目签名打包成apk,结果打包的时候出错了,吃惊,什么情况.等成功打包以后,安装起来发现部分功能又报错了,囧,所幸最后还是解决了.在这里记录一下遇 ...
HTML与XML关系分析
本来这篇是为CSS准备的,但看到视频中CSS和HTML.XML都有关系,即,都是设置他们的样式.而XML和HTML的格式看着也有些类似,就不得不分析一下二者之间的关系了. 要想分析事物关系,要先弄清他 ...
document&period;compatMode属性介绍
之前不了解这个属性,今天总结一下,以后可能会用到. 对于document.compatMode,很多朋友可能都根我一样很少接触,知道他的存在却不清楚他的用途.今天在ext中看到 document.co ...
FFmpeg开发实战（四）：FFmpeg 抽取音视频的音频数据
如何使用FFmpeg抽取音视频的音频数据,代码如下: void adts_header(char *szAdtsHeader, int dataLen); // 使用FFmpeg从视频中抽取音频 vo ...
Win10系列：C&num;应用控件基础20
SemanticZoom控件 SemanticZoom控件由相互关联的缩小视图和放大视图所组成,缩小视图用来显示内容的索引,放大视图可以用来显示内容的详细信息,用户可以根据阅读需要在两种视图之间*切 ...
Numpy中stack()，hstack()，vstack()函数详解
一`.stack 按指定维度堆叠数组. stack(a, b) 维度计算 axis=0: 2*m*n axis=1: m*2*n axis=-1: m*n*2 a = np.arange( ...