Java线程池Executors分析

Posted on 2018-11-23 Edited on 2024-12-07 In juc

简介

Executors 这个类，因为它仅仅是工具类，它的所有方法都是 static 的。

Demo

package com.fly.learn.executor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 线程池test
 * @author: peijiepang
 * @date 2018/11/23
 * @Description:
 */
public class ThreadPoolExecutorTest {
    private final static Logger LOGGER = LoggerFactory.getLogger(ThreadPoolExecutorTest.class);
    //private static ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
    //private static Executor executor = Executors.newFixedThreadPool(5);
    //private static Executor executor = Executors.newSingleThreadExecutor();
    //private static Executor executor = Executors.newCachedThreadPool();
    private static ExecutorService executor = Executors.newScheduledThreadPool(5);

    public void executeTask(){
        Task1 task1 = new Task1();//构建任务
        executor.execute(task1);//执行任务
    }

    /*
     * 基本任务2
     */
    class Task1 implements Runnable{
        public void run() {
            //具体任务的业务
            for(int i=0;i<1000;i++){
                LOGGER.info("{}...",i);
            }
        }
    }
    public static void main(String[] args) {
        ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();
        test.executeTask();
        executor.shutdown();
        while (!executor.isTerminated()){
            LOGGER.info("finish......");
        }
    }
}

Executors可以创建的几种线程

newFixedThreadPool(int corePoolSize)
- 创建一个线程数固定（corePoolSize==maximumPoolSize）的线程池
- 核心线程会一直运行
- 如果一个核心线程由于异常挂了，会新创建一个线程
- 无界队列LinkedBlockingQueue
newSingleThreadExecutor
- 创建一个线程数固定（corePoolSize==maximumPoolSize==1）的线程池
- 核心线程会一直运行
- 无界队列LinkedBlockingQueue
- 所有task都是串行执行的（即同一时刻只有一个任务在执行）
newCachedThreadPool
- corePoolSize==0
- maximumPoolSize==Integer.MAX_VALUE
- 队列：SynchronousQueue
- 创建一个线程池：当池中的线程都处于忙碌状态时，会立即新建一个线程来处理新来的任务
- 这种池将会在执行许多耗时短的异步任务的时候提高程序的性能
- 60秒内没有使用的线程将会被中止，并且从线程池中移除，因此几乎不必担心耗费资源
newScheduledThreadPool(int corePoolSize)
- 用于执行定时或延迟执行的任务，最典型的：异步操作时的超时回调

源码分析

/**
 * 1、创建一个线程数固定（corePoolSize==maximumPoolSize）的线程池,
 * 2、核心线程会一直运行
 * 3、无界队列LinkedBlockingQueue
 */
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

/**
 * 用于创建ForkJoin框架中用到的ForkJoinPool线程
 * @param parallelism 并行数
 * @return
 */
public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

public static ExecutorService newWorkStealingPool() {
    return new ForkJoinPool
        (Runtime.getRuntime().availableProcessors(),
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

/**
 * 1、创建一个线程数固定（corePoolSize==maximumPoolSize==1）的线程池
 * 2、核心线程会一直运行
 * 3、无界队列LinkedBlockingQueue
 * 注意：所有task都是串行执行的
 */
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}

/**
 * 1、创建一个线程池：当池中的线程都处于忙碌状态时，会立即新建一个线程来处理新来的任务
 * 2、这种池将会在执行许多耗时短的异步任务的时候提高程序的性能。
 * 3、60秒内没有使用的线程将会被中止，并且从线程池中移除，因此几乎不必担心耗费资源
 * 4、队列：SynchronousQueue
 * 5、maximumPoolSize为Integer.MAX_VALUE
 */
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

/**
 * 创建一个线程池：该线程池可以用于执行延时任务或者定时任务
 */
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

/**
 * 主要用于包装现有的线程池，包装之后的线程池不能修改
 * @param executor
 * @return
 */
public static ExecutorService unconfigurableExecutorService(ExecutorService executor) {
    if (executor == null)
        throw new NullPointerException();
    return new DelegatedExecutorService(executor);
}

/**
 * 用于包装可以周期性执行任务的线程池，包装之后的线程池不能修改
 * @param executor
 * @return
 */
public static ScheduledExecutorService unconfigurableScheduledExecutorService(ScheduledExecutorService executor) {
    if (executor == null)
        throw new NullPointerException();
    return new DelegatedScheduledExecutorService(executor);
}

/**
 * 默认的工厂方法类
 * @return
 */
public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}