线程池概述

Posted 2020-11-16 loveer

线程池的好处

节省创建和销毁线程的巨大开销

不用等待创建线程而延迟任务，提高系统响应性

控制线程数量，防止内存耗尽

Executor的生命周期

Executor扩展了ExecutorService接口，用于管理生命周期

ExecutorService生命周期有三种状态：
    运行
    关闭
    已终止

关闭ExecutorService：
    exec.shutdown();
    exec.awaitTermination(long timeout, TimeUnit unit);

public interface ExecutorService extends Executor {

    void shutdown();    //不再接受新任务，同时等待已提交的任务执行完成

    List<Runnable> shutdownNow();   //不再接受新任务，并尝试取消正在运行的任务

    boolean isShutdown();

    boolean isTerminated(); //轮询ExecutorService是否已经终止

    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

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

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);


    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

管理列队任务

newFixedThreadPool 和 newSingleThreadExecutor 默认使用无界的 LinkedBlockingQueue。

一种更稳妥的资源管理策略是使用有界列队：
    AarrayBlockingQueue
    有界的 LinkedBlockingQueue , PriorityBlockingQueue


当队列满时，新的任务需要饱和策略（Saturation Policy）解决问题。


对于非常大的或者无界的线程池，可以使用 SynchronousQueue 避免任务排队，直接将任务从生产者移交给工作者线程：

    如果将一个元素放入 SynchronousQueue 中，必须有另一个线程正在等待接受这个元素，否则将创建新的线程。
    如果线程池的当前大小达到最大值，则根据拒绝策略拒绝该任务。

    只有当线程是无界的或者可以拒绝任务时（根据饱和策略）, SynchronousQueue 才有实际价值，比如 newCachedThreadPool。

饱和策略

AbortPolicy（默认）：
    抛出未检查异常（RejectedExecutionException），调用者可以捕获异常。

DiscardPolicy：
    悄悄抛弃该任务

DiscardOldestPolicy：
    抛弃下一个将被执行的任务，重新尝试提交新的任务。

CallerRunsPolicy：
    将任务回退到调用者，不会在线程池的某个线程中执行，在一个调用了 execute 的线程（主线程）中执行该任务。
    由于执行任务需要时间，则主线程在一定时间里无法提交任务。


/**
 * 使用信号量阻塞提交任务的线程，控制任务的提交速率
 */
public class BoundedExecutor {
    private final Executor executor;
    private final Semaphore semaphore;

    public BoundedExecutor(Executor executor, int bound) {
        this.executor = executor;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable task) throws InterruptedException {
        semaphore.acquire();
        
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        task.run();
                    }finally {
                        semaphore.release();
                    }
                }
            });
        }catch (RejectedExecutionException e){
            semaphore.release();
        }
    }
}

线程工厂

通过指定一个线程工厂方法，可以定制线程池的配置信息。

默认的线程工厂方法将创建一个新的，非守护的线程：

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }