【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提供了的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架，该类在有递归实现的场景有更优异的表现。

package com.concurrent.test;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;/*** 测试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 submitRunable() throws InterruptedException, ExecutionException {ExecutorService executorService = Executors.newSingleThreadExecutor();Future<?> future = executorService.submit(new Runnable() {@Overridepublic 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() {@Overridepublic void run() {System.out.println("This is submitRunnableWithResult");}}, RESULT);return future.get();}private static Integer submitBlockCallable() throws InterruptedException, ExecutionException {ExecutorService executorService = Executors.newSingleThreadExecutor();Future<Integer> future = executorService.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {System.out.println("This is submitBlockCallable");return RESULT;}});return future.get(); //阻塞
    }private static boolean submitNonBlockCallable() {ExecutorService executorService = Executors.newSingleThreadExecutor();Future<Integer> future = executorService.submit(new Callable<Integer>() {@Overridepublic 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() {ExecutorService executorService = Executors.newSingleThreadExecutor();final StringBuilder sb = new StringBuilder();executorService.submit(new Callable<Integer>() {@Overridepublic 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 {ExecutorService executorService = Executors.newSingleThreadExecutor();final StringBuilder sb = new StringBuilder();executorService.submit(new Callable<Integer>() {@Overridepublic 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();}@Testpublic void test() throws InterruptedException, ExecutionException {Assert.assertTrue(submitRunable());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类如下：

package com.concurrent.test;import java.util.List;
import java.util.concurrent.RecursiveTask;public class SumTask extends RecursiveTask<Integer> {/*** */private static final long serialVersionUID = -5468389855825594053L;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*/@Overrideprotected 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) {int sum = 0;for (Integer integer : list) {sum += integer;}return sum;}}

转载于:https://www.cnblogs.com/zhaoyibing/p/9628759.html

【Java多线程系列七】ExecutorService相关推荐

Java多线程系列七——ExecutorService
java.util.concurrent.ExecutorService接口提供了许多线程管理的方法 Method 说明 shutdown 拒绝接收新的任务,待已提交的任务执行后关闭,且宿主线程不阻塞 ...
Java多线程系列(七)：并发容器的原理，7大并发容器详解、及使用场景
之前谈过高并发编程系列: 高并发编程系列:4种常用Java线程锁的特点,性能比较.使用场景高并发编程系列:CountDownLatch.Semaphore等4大并发工具类详解高并发编程系列:4大J ...
Java多线程系列(六)：深入详解Synchronized同步锁的底层实现
谈到多线程就不得不谈到Synchronized,很多同学只会使用,缺不是很明白整个Synchronized的底层实现原理,这也是面试经常被问到的环节,比如: synchronized的底层实现原理 s ...
Java多线程系列(九)：CountDownLatch、Semaphore等4大并发工具类详解
之前谈过高并发编程系列:4种常用Java线程锁的特点,性能比较.使用场景 ,以及高并发编程系列:ConcurrentHashMap的实现原理(JDK1.7和JDK1.8) 今天主要介绍concurre ...
Java多线程系列(二)：线程的五大状态，以及线程之间的通信与协作
在Java面试的时候,经常会问到Java并发编程相关的多线程.线程池.线程锁.线程通信等面试必考点,比如: Java并发编程系列:Java线程池的使用方式,核心运行原理.以及注意事项 Java并发编程 ...
Java多线程系列(五)：线程池的实现原理、优点与风险、以及四种线程池实现
为什么需要线程池我们有两种常见的创建线程的方法,一种是继承Thread类,一种是实现Runnable的接口,Thread类其实也是实现了Runnable接口.但是我们创建这两种线程在运行结束后都会被 ...
Java多线程系列(十)：源码剖析AQS的实现原理
在并发编程领域,AQS号称是并发同步组件的基石,很多并发同步组件都是基于AQS实现,所以想掌握好高并发编程,你需要掌握好AQS. 本篇主要通过对AQS的实现原理.数据模型.资源共享方式.获取锁的过程, ...
Java 多线程（七）线程间的通信
Java 多线程(七) 线程间的通信--wait及notify方法线程间的相互作用线程间的相互作用:线程之间需要一些协调通信,来共同完成一件任务. Object类中相关的方法有两个notify方法 ...
Java多线程系列--“JUC线程池”06之 Callable和Future
转载自 Java多线程系列--"JUC线程池"06之 Callable和Future Callable 和 Future 简介 Callable 和 Future 是比较有趣的一 ...

【Java多线程系列七】ExecutorService

【Java多线程系列七】ExecutorService相关推荐

最新文章

热门文章