Java高并发-线程池

发表于 | 分类于
字数统计: 2.5k 字 | 阅读时长 ≈ 9 分钟

Executor

控制线程的启动、执行和关闭，可以简化并发编程的操作。

通过Executor来启动线程比使用Thread的start方法更好，除了更易管理，效率更好（用线程池实现，节约开销）外，还有关键的一点：有助于避免this逃逸问题。

Executor接口中之定义了一个方法execute（Runnable command），该方法接收一个Runable实例，它用来执行一个任务，任务即一个实现了Runnable接口的类。

ExecutorService

ExecutorService接口继承自Executor接口，它提供了更丰富的实现多线程的方法：

1
2
3
Future<?> submit(Runnable task)

<T> Future<T> submit(Callable<T> task)

ExecutorService的生命周期包括三种状态：运行、关闭、终止。创建后便进入运行状态，当调用了shutdown（）方法时，便进入关闭状态，此时意味着ExecutorService不再接受新的任务，但它还在执行已经提交了的任务，当素有已经提交了的任务执行完后，便到达终止状态。如果不调用shutdown（）方法，ExecutorService会一直处在运行状态，不断接收新的任务，执行新的任务，服务器端一般不需要关闭它，保持一直运行即可。

submit方便Exception处理

Callable

Callable任务有返回值。并且Callable的call()方法只能通过ExecutorService的submit(Callable task) 方法来执行，并且返回一个 Future，是表示任务等待完成的 Future。

如果Future的返回尚未完成，则get（）方法会阻塞等待，直到Future完成返回，可以通过调用isDone（）方法判断Future是否完成了返回。

Executors

Executors提供了一系列工厂方法用于创先线程池，返回的线程池都实现了ExecutorService接口。

ThreadPoolExecutor

自定义线程池，可以用ThreadPoolExecutor类创建，它有多个构造方法来创建线程池，用该类很容易实现自定义的线程池

1
public ThreadPoolExecutor (int corePoolSize, int maximumPoolSize, long         keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue)

corePoolSize：线程池中所保存的核心线程数，包括空闲线程。

maximumPoolSize：池中允许的最大线程数。

keepAliveTime：线程池中的空闲线程所能持续的最长时间。

unit：持续时间的单位。

workQueue：任务执行前保存任务的队列，仅保存由execute方法提交的Runnable任务。

根据ThreadPoolExecutor源码前面大段的注释，我们可以看出，当试图通过excute方法讲一个Runnable任务添加到线程池中时，按照如下顺序来处理：

1. 如果线程池中的线程数量少于corePoolSize，即使线程池中有空闲线程，也会创建一个新的线程来执行新添加的任务；

2. 如果线程池中的线程数量大于等于corePoolSize，但缓冲队列workQueue未满，则将新添加的任务放到workQueue中，按照FIFO的原则依次等待执行（线程池中有线程空闲出来后依次将缓冲队列中的任务交付给空闲的线程执行）；

3. 如果线程池中的线程数量大于等于corePoolSize，且缓冲队列workQueue已满，但线程池中的线程数量小于maximumPoolSize，则会创建新的线程来处理被添加的任务；

4. 如果线程池中的线程数量等于了maximumPoolSize，有4种才处理方式

Future

Future 表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。

get方法
等待计算完成，然后获取其结果。阻塞

isDone方法
用来查询任务是否做完，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/*新建一个Callable任务*/
Callable<Integer> callableTask = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        System.out.println("Calculating 1+1!");
        TimeUnit.SECONDS.sleep(2);//休眠2秒
        return 2;
    }
};
ExecutorService executor = Executors.newCachedThreadPool();
Future<Integer> result = executor.submit(callableTask);
executor.shutdown();
while(!result.isDone()){//isDone()方法可以查询子线程是否做完
    System.out.println("子线程正在执行");
    TimeUnit.SECONDS.sleep(1);//休眠1秒
}
try {
    System.out.println("子线程执行结果:"+result.get());
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

FutureTask

FutureTask类是 Future 接口的一个实现。FutureTask类实现了RunnableFuture接口，RunnableFuture继承了Runnable接口和Future接口，所以：

FutureTask可以作为Runnable被线程执行

可以作为Future得到传入的Callable对象的返回值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
FutureTask<Integer> futureTask = new FutureTask<>(new Callable<Integer>() {
   @Override
   public Integer call() throws Exception {
        System.out.println("futureTask is wokring 1+1!");
        return 2;
   }
});
Thread t1 = new Thread(futureTask);//1.可以作为Runnable类型对象使用
t1.start();
try {
   System.out.println(futureTask.get());//2.可以作为Future类型对象得到线程运算返回值
} catch (ExecutionException e) {
   e.printStackTrace();
}

newFixedThreadPool

创建固定数目线程的线程池。

  • newFixedThreadPool与cacheThreadPool差不多，也是能reuse就用，但不能随时建新的线程
  • 其独特之处:任意时间点，最多只能有固定数目的活动线程存在，此时如果有新的线程要建立，只能放在另外的队列中等待，直到当前的线程中某个线程终止直接被移出池子
  • 和cacheThreadPool不同，FixedThreadPool没有IDLE机制（可能也有，但既然文档没提，肯定非常长，类似依赖上层的TCP或UDP IDLE机制之类的），所以FixedThreadPool多数针对一些很稳定很固定的正规并发线程，多用于服务器
  • 从方法的源代码看，cache池和fixed 池调用的是同一个底层 池，只不过参数不同:
    fixed池线程数固定，并且是0秒IDLE（无IDLE）
    cache池线程数支持0-Integer.MAX_VALUE(显然完全没考虑主机的资源承受能力），60秒IDLE

newCachedThreadPool

创建一个可缓存的线程池，调用execute将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线 程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。

  • 缓存型池子，先查看池中有没有以前建立的线程，如果有，就 reuse.如果没有，就建一个新的线程加入池中
  • 缓存型池子通常用于执行一些生存期很短的异步型任务
    因此在一些面向连接的daemon型SERVER中用得不多。但对于生存期短的异步任务，它是Executor的首选。

  • 能reuse的线程，必须是timeout IDLE内的池中线程，缺省 timeout是60s,超过这个IDLE时长，线程实例将被终止及移出池。

    注意，放入CachedThreadPool的线程不必担心其结束，超过TIMEOUT不活动，其会自动被终止。

newSingleThreadExecutor

创建一个单线程化的Executor。

  • 单例线程，任意时间池中只能有一个线程
  • 用的是和cache池和fixed池相同的底层池，但线程数目是1-1,0秒IDLE（无IDLE）

newScheduledThreadPool

创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。

  • 调度型线程池
  • 这个池子里的线程可以按schedule依次delay执行，或周期执行

newWorkStealingPool

该线程池实际上是一个精灵线程(deamon,守护线程、后台线程)池，如果主线程不阻塞的话，精灵线程池中定义的线程是不会执行的，只有当主线程阻塞时候，才会触发精灵线程池中的全部线程执行。

当前线程池工作完成后，会帮助其他未完成的线程池工作

1
2
3
4
ExecutorService service = Executors.newWorkStealingPool();
service.execute(...);//提交任务
System.in.read();
//主线程在这里阻塞,会触发上面精灵线程池中的线程，执行在Runnable接口中实现的任务。

ForkJoinPool

ForkJoinTask代表一个需要执行的任务，真正执行这些任务的线程是放在一个线程池(ForkJoinPool)里面。ForkJoinPool是一个可以执行ForkJoinTask的ExcuteService，与ExcuteService不同的是它采用了work-stealing模式：

所有在池中的线程尝试去执行其他线程创建的子任务，这样就很少有线程处于空闲状态，非常高效。

其他

下面说说几种排队的策略：

1、直接提交。缓冲队列采用 SynchronousQueue，它将任务直接交给线程处理而不保持它们。如果不存在可用于立即运行任务的线程（即线程池中的线程都在工作），则试图把任务加入缓冲队列将会失败，因此会构造一个新的线程来处理新添加的任务，并将其加入到线程池中。直接提交通常要求无界 maximumPoolSizes（Integer.MAX_VALUE） 以避免拒绝新提交的任务。newCachedThreadPool采用的便是这种策略。

2、无界队列。使用无界队列（典型的便是采用预定义容量的 LinkedBlockingQueue，理论上是该缓冲队列可以对无限多的任务排队）将导致在所有 corePoolSize 线程都工作的情况下将新任务加入到缓冲队列中。这样，创建的线程就不会超过 corePoolSize，也因此，maximumPoolSize 的值也就无效了。当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列。newFixedThreadPool采用的便是这种策略。

3、有界队列。当使用有限的 maximumPoolSizes 时，有界队列（一般缓冲队列使用ArrayBlockingQueue，并制定队列的最大长度）有助于防止资源耗尽，但是可能较难调整和控制，队列大小和最大池大小需要相互折衷，需要设定合理的参数。

对于最后一个拒绝策略参数，通常解决方式是：

  1. AbortPolicy：直接抛出异常，系统继续工作，默认策略。
  2. CallerRunsPolicy：只要线程池未关闭，该策略直接在调用者线程中执行，运行当前被丢弃的任务。
  3. DiscardOrderstPolicy：丢弃最老的请求，尝试再次提交当前任务。
  4. 丢弃无法处理的任务，不给于任何处理。

如果需要自定义策略，需要实现RejectedExecutionHandler接口。

-------------本文结束感谢您的阅读-------------
Dean Wang wechat