Java 线程池工具类

1. 线程池工具类（Executors）

• 多数都是静态方法

public class Executors {
    public static ExecutorService newCachedThreadPool() {}
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {}

    public static ExecutorService newFixedThreadPool(int nThreads) {}
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {}
   
}

1.1 ThreadFactory

创建线程池时，默认用的是 Executors.defaultThreadFactory()

    public static ThreadFactory defaultThreadFactory() {
        return new DefaultThreadFactory();
    }

2. 创建线程池

2.1 newCachedThreadPool

• keepAliveTime: 60
• SynchronousQueue

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

2.2 newFixedThreadPool

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

2.3 newSingleThreadExecutor

• 单一线程池可以实现以队列的方式来执行任务

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

2.4 newScheduledThreadPool

• 定时任务（作心跳检测，Eureka/Nacos/Dubbo）

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

public interface ScheduledExecutorService extends ExecutorService {
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
}

2.5 newWorkStealingPool

• Fork/Join(MapReduce)

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

3. 线程池工具类的使用

public void demo() {
    ExecutorService executorService = Executors.newCachedThreadPool();
    executorService.execute(() -> {});
    executorService.execute(() -> {});
    executorService.shutdown();
}

public void poolDemo() {
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                // pool-1-thread-1: do something...
                System.out.println(Thread.currentThread().getName() + ": do something...");
            }
        });
    }
    executorService.shutdown();
}

线程池中线程对象只有处于闲置状态时，才可以被复用

public void demo() throws InterruptedException {
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 5; i++) {
        executorService.execute(new MyRunnable(String.valueOf(i)));
    }
    TimeUnit.MILLISECONDS.sleep(10);
    for (int i = 0; i < 5; i++) {
        executorService.execute(new MyRunnable(String.valueOf(i)));
    }
}

public class MyRunnable implements Runnable {
    private String name;
    public MyRunnable(String name) {
        this.name = name;
    }
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + ": " + name);
    }
}

3.2 定时任务

public void schedule() throws ExecutionException, InterruptedException {
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);

    for (int i = 0; i < 10; i++) {
        Future<Integer> result = pool.schedule(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int num = new Random().nextInt(100);
                System.out.println(Thread.currentThread().getName() + " : " + num);
                return num;
            }
        }, 1, TimeUnit.SECONDS);
        System.out.println(result.get());
    }
    pool.shutdown();
}

4. 定时器创建线程

public static void main(String[] args) {
    Timer timer = new Timer();
    timer.scheduleAtFixedRate(new TimerTask() {
        @Override
        public void run() {
            log.info(Thread.currentThread().getName());
        }
    }, 1000, 1000);
}