多线程之线程池基础

原创
2020/03/12 16:44
阅读数 153

优点:

  1. 避免线程的创建和销毁带来的性能开销。
  2. 避免大量的线程间因互相抢占系统资源导致的阻塞现象。
  3. 能够对线程进行简单的管理并提供定时执行、间隔执行等功能。

概念:

        java通过Executors类提供了四种常用线程池。不过线程池的真正顶级接口是ExecutorService,继承了Executor接口。Executors中ExecutorService使用ThreadPoolExecutor提供线程,而ThreadPoolExecutor也最终继承至ExecutorService。

使用:

newCachedThreadPool()

概念:

        java中使用newCachedThreadPool()可以创建一个可缓存线程池,其中线程的数量不固定,可以根据需要自由的创建或回收线程。可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同(例如超时参数)的线程池。

用法:

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();

        for (int i = 0; i < 5; i++) {
            final int index = i;
            executorService.execute(new Runnable() {
                public void run() {
                    try {
                        SimpleDateFormat sdf = new SimpleDateFormat(
                                "HH:mm:ss");
                        System.out.println("运行时间: " +
                                sdf.format(new Date()) + " " + index + " " + Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        executorService.shutdown();
    }
}
运行时间: 11:22:08 4 pool-1-thread-5
运行时间: 11:22:08 0 pool-1-thread-1
运行时间: 11:22:08 3 pool-1-thread-4
运行时间: 11:22:08 1 pool-1-thread-2
运行时间: 11:22:08 2 pool-1-thread-3

        newCachedThreadPool()创建的线程池每次有任务来临时就会创建新的线程来执行任务,因此循环到来5个任务,会创建5个线程处理。可以看出5个线程是同时运行的。

newSingleThreadExecutor()

概念:

        java中使用newCachedThreadPool()可以创建一个单线程的线程池。所有的任务使用同一个线程来处理任务,所有的任务会按照指定的顺序执行。

用法:

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

        for (int i = 0; i < 5; i++) {
            final int index = i;
            executorService.execute(new Runnable() {
                public void run() {
                    try {
                        SimpleDateFormat sdf = new SimpleDateFormat(
                                "HH:mm:ss");
                        System.out.println("运行时间: " +
                                sdf.format(new Date()) + " " + index + "" + Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        executorService.shutdown();
运行时间: 11:20:47 0 pool-1-thread-1
运行时间: 11:20:49 1 pool-1-thread-1
运行时间: 11:20:51 2 pool-1-thread-1
运行时间: 11:20:53 3 pool-1-thread-1
运行时间: 11:20:55 4 pool-1-thread-1

        可以看出同时到来的5个任务使用相同的线程处理任务。每个任务执行完成后下个任务才能执行。

newFixedThreadPool()

概念:

        java中使用newCachedThreadPool()可以创建一个固定大小的线程池。每到来一个任务时,就会创建一个线程来处理,直到线程池满,不再增大。如果某个线程因异常而结束,会补充一个线程进来。

用法:

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

        for (int i = 0; i < 5; i++) {
            final int index = i;
            executorService.execute(new Runnable() {
                public void run() {
                    try {
                        SimpleDateFormat sdf = new SimpleDateFormat(
                                "HH:mm:ss");
                        System.out.println("运行时间: " +
                                sdf.format(new Date()) + " " + index + " " + Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        executorService.shutdown();
运行时间: 11:34:18 1 pool-1-thread-2
运行时间: 11:34:18 2 pool-1-thread-3
运行时间: 11:34:18 0 pool-1-thread-1
运行时间: 11:34:20 3 pool-1-thread-3
运行时间: 11:34:20 4 pool-1-thread-2

        可以看到,线程池在前三个任务到来时,创建了三个线程,第四个任务来临时进入等待队列,需要等待前面的任务执行完成,待前面的任务执行完成后,等待队列中的任务才可以继续执行,

newScheduledThreadPool()

概念:

会创建一个固定大小的线程池,该线程池支持周期性执行和定时任务。

用法:

定时执行:

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
        SimpleDateFormat sdf = new SimpleDateFormat(
                "HH:mm:ss");
        for (int i = 0; i < 5; i++) {
            final int index = i;
            System.out.println("提交时间: " + sdf.format(new Date()));
            executorService.schedule(new Runnable() {
                public void run() {
                    try {

                        System.out.println("运行时间: " +
                                sdf.format(new Date()) + " " + index + " " + Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, 3, TimeUnit.SECONDS);

        }

        executorService.shutdown();
提交时间: 14:05:46
提交时间: 14:05:46
提交时间: 14:05:46
提交时间: 14:05:46
提交时间: 14:05:46
运行时间: 14:05:49 0 pool-1-thread-1
运行时间: 14:05:49 1 pool-1-thread-3
运行时间: 14:05:49 2 pool-1-thread-2
运行时间: 14:05:51 3 pool-1-thread-2
运行时间: 14:05:51 4 pool-1-thread-3

        可以看出,在任务提交3秒钟后,任务才开始执行,并且线程池达到最大之后,多余队列中的任务就只能等到有线程空闲了才能继续执行。

周期执行:

ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
        SimpleDateFormat sdf = new SimpleDateFormat(
                "HH:mm:ss");
//        for (int i = 0; i < 5; i++) {
            final int index = 1;
            System.out.println("提交时间: " + sdf.format(new Date()));
            executorService.scheduleAtFixedRate(new Runnable() {
                public void run() {
                    try {

                        System.out.println("运行时间: " +
                                sdf.format(new Date()) + " " + index + " " + Thread.currentThread().getName());
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, 1, 3, TimeUnit.SECONDS);
            //主线程等待10秒钟后关闭
            Thread.sleep(10000);
//        }

        executorService.shutdown();
提交时间: 14:29:49
运行时间: 14:29:50 1 pool-1-thread-1
运行时间: 14:29:53 1 pool-1-thread-1
运行时间: 14:29:56 1 pool-1-thread-2
运行时间: 14:29:59 1 pool-1-thread-2

        可以看出在主线程提交任务1秒后,子线程开始执行,子线程每隔3秒,周期性执行一次。

总结:

根据前面的可以看出线程池最终使用两种方法创建线程。

工厂方法 实现类 任务队列
newCachedThreadPool ThreadPoolExecutor SynchronousQueue
newSingleThreadExecutor ThreadPoolExecutor LinkedBlockingQueue
newFixedThreadPool ThreadPoolExecutor LinkedBlockingQueue
newScheduledThreadPool ScheduledThreadPoolExecutor DelayedWorkQueue

分别使用ThreadPoolExecutor()方法和ScheduledThreadPoolExecutor()方法。

首先看ThreadPoolExecutor()方法:

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

看参数:

corePoolSize:要保留在池中的线程数,即使它们处于空闲状态,除非设置了 {@code 允许 CoreThreadTimeOut}。核心池大小(最大最小没搞懂),一般来说不管有没有任务都会在池中存活。allowCoreThreadTimeOut设置为true时,核心线程超过固定时间则会被销毁。其中corePoolSize中的线程并不是建立线程池时立即创建的,而是在任务到来时创建的。

maximumPoolSize:线程池中线程总数的最大值,其线程总数等于核心线程数加非核心线程数。非核心线程就是线程数量超过核心线程数后建立的线程。

keepAliveTime:非核心线程在线程池中存活的时间,当线程数大于核心时,这是多余的空闲线程在终止之前等待新任务的最大时间。

unit:存活时间的时间单位。

workQueue:用于在执行任务之前保存任务的队列。 此队列将仅保留 [@code 执行] 方法提交的 [@code 可运行] 任务。任务队列,核心线程都在运行时,用来维护新到来的任务,该队列满时才会新建非核心线程处理。核心线程>任务队列>非核心线程。(SynchronousQueue,LinkedBlockingQueue,ArrayBlockingQueue,DelayQueue

threadFactory:一个接口,线程工厂,用来为线程池创建线程。

handler:用来处理异常情况。

任务处理过程:

调用核心线程处理->放入任务队列->创建非核心线程->拒绝任务

问题:

        在newSingleThreadExecutor与newFixedThreadPool中都使用了LinkedBlockingQueue来充当任务队列,理论上可以无限加入任务。但是,允许的请求队列长度为Integer.MAX_VALUE。可能会堆积大量请求,造成内存泄露。

        在newCachedThreadPool与newScheduledThreadPool中。允许的创建的线程数量为Integer.MAX_VALUE,造成创建大量的线程(毕竟每个线程创建都需要分配内存),造成内存泄露。

解决:

阿里开发手册建议使用ThreadPoolExecutor自定义创建线程池。执行思路与上面的任务处理过程相同。

扩展:

workQueue:任务等待队列。

  1. ArrayBlockingQueue:由数组构成的有界阻塞队列。正常流程超过maximumPoolSize会出错。
  2. LinkedBlockingQueue:由链表构成的有界阻塞队列。该队列一般不设置大小,maximumPoolSize会失效,会导致任务无限进入队列等候,BUT,OOM来了。
  3. PriorityBlockingQueue:支持优先级排序的无界阻塞队列。
  4. DelayQueue:使用优先级队列实现的无界阻塞队列。
  5. SynchronousQueue:不存储元素的无界队列。该队列接受到任务会直接交给线程处理,如果线程已满则新建线程处理。
  6. LinkedTransferQueue:由链表构成的无界阻塞队列。
  7. LinkedBlockingDeque:由链表构成的双向阻塞队列。

handler:任务拒绝策略。

  1. AbortPolicy:默认策略,丢弃任务,抛出 RejectedExecutionException 异常。
  2. CallerRunsPolicy:由调用线程(当前线程池所在线程,父线程)处理该任务。
  3. DiscardOldestPolicy:丢弃队列前面的任务,将此任务加入队列。
  4. DiscardPolicy:丢弃任务,不抛出异常。
展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部