• 技术文章 >Java >java教程

    Java中ThreadPoolExecutor线程池的用法介绍

    不言不言2019-04-04 09:55:36转载1482

    本篇文章给大家带来的内容是关于Java中ThreadPoolExecutor线程池的用法介绍,有一定的参考价值,有需要的朋友可以参考一下,希望对你有所帮助。

    Executors

    Executors 是一个Java中的工具类. 提供工厂方法来创建不同类型的线程池.

    3577595468-5ca357e1bca08_articlex.png

    从上图中也可以看出, Executors的创建线程池的方法, 创建出来的线程池都实现了 ExecutorService接口. 常用方法有以下几个:

    newFixedThreadPool(int Threads): 创建固定数目线程的线程池, 超出的线程会在队列中等待.

    newCachedThreadPool(): 创建一个可缓存线程池, 如果线程池长度超过处理需要, 可灵活回收空闲线程(60秒), 若无可回收,则新建线程.

    newSingleThreadExecutor(): 创建一个单线程化的线程池, 它只会用唯一的工作线程来执行任务, 保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行. 如果某一个任务执行出错, 将有另一个线程来继续执行.

    newScheduledThreadPool(int corePoolSize): 创建一个支持定时及周期性的任务执行的线程池, 多数情况下可用来替代Timer类.

    Executors 例子

    newCachedThreadPool

    线程最大数为 Integer.MAX_VALUE, 当我们往线程池添加了 n 个任务, 这 n 个任务都是一起执行的.

            ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            Thread.currentThread().sleep(1000);
                            System.out.println(Thread.currentThread().getName());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
    
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            Thread.currentThread().sleep(1000);
                            System.out.println(Thread.currentThread().getName());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

    newFixedThreadPool

            ExecutorService cachedThreadPool = Executors.newFixedThreadPool(1);
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            Thread.currentThread().sleep(1000);
                            System.out.println(Thread.currentThread().getName());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
    
            cachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            Thread.currentThread().sleep(1000);
                            System.out.println(Thread.currentThread().getName());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

    newScheduledThreadPool

    三秒执行一次, 只有执行完这一次后, 才会执行.

            ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
            scheduledExecutorService.schedule(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            Thread.currentThread().sleep(2000);
                            System.out.println(Thread.currentThread().getName());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }, 3, TimeUnit.SECONDS);

    newSingleThreadExecutor

    顺序执行各个任务, 第一个任务执行完, 才会执行下一个.

            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            System.out.println(Thread.currentThread().getName());
                            Thread.currentThread().sleep(10000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
    
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            System.out.println(Thread.currentThread().getName());
                            Thread.currentThread().sleep(2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

    Executors存在什么问题

    1195103511-5ca4248a8b912_articlex.jpg

    在阿里巴巴Java开发手册中提到,使用Executors创建线程池可能会导致OOM(OutOfMemory ,内存溢出),但是并没有说明为什么,那么接下来我们就来看一下到底为什么不允许使用Executors?

    我们先来一个简单的例子,模拟一下使用Executors导致OOM的情况.

    /**
     * @author Hollis
     */
    public class ExecutorsDemo {
        private static ExecutorService executor = Executors.newFixedThreadPool(15);
        public static void main(String[] args) {
            for (int i = 0; i < Integer.MAX_VALUE; i++) {
                executor.execute(new SubThread());
            }
        }
    }
    
    class SubThread implements Runnable {
        @Override
        public void run() {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                //do nothing
            }
        }
    }

    通过指定JVM参数:-Xmx8m -Xms8m 运行以上代码,会抛出OOM:

    Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
        at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
        at com.hollis.ExecutorsDemo.main(ExecutorsDemo.java:16)

    以上代码指出,ExecutorsDemo.java 的第16行,就是代码中的 executor.execute(new SubThread());

    Java中的 BlockingQueue 主要有两种实现, 分别是 ArrayBlockingQueue 和 LinkedBlockingQueue.

    ArrayBlockingQueue 是一个用数组实现的有界阻塞队列, 必须设置容量.

    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    LinkedBlockingQueue 是一个用链表实现的有界阻塞队列, 容量可以选择进行设置, 不设置的话, 将是一个无边界的阻塞队列, 最大长度为 Integer.MAX_VALUE.

    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

    这里的问题就出在如果我们不设置 LinkedBlockingQueue 的容量的话, 其默认容量将会是 Integer.MAX_VALUE.

    而 newFixedThreadPool 中创建 LinkedBlockingQueue 时, 并未指定容量. 此时, LinkedBlockingQueue 就是一个无边界队列, 对于一个无边界队列来说, 是可以不断的向队列中加入任务的, 这种情况下就有可能因为任务过多而导致内存溢出问题.

    newCachedThreadPool 和 newScheduledThreadPool 这两种方式创建的最大线程数可能是Integer.MAX_VALUE, 而创建这么多线程, 必然就有可能导致OOM.

    ThreadPoolExecutor 创建线程池

    避免使用 Executors 创建线程池, 主要是避免使用其中的默认实现, 那么我们可以自己直接调用 ThreadPoolExecutor 的构造函数来自己创建线程池. 在创建的同时, 给 BlockQueue 指定容量就可以了.

    ExecutorService executor = new ThreadPoolExecutor(10, 10,
            60L, TimeUnit.SECONDS,
            new ArrayBlockingQueue(10));

    这种情况下, 一旦提交的线程数超过当前可用线程数时, 就会抛出 java.util.concurrent.RejectedExecutionException, 这是因为当前线程池使用的队列是有边界队列, 队列已经满了便无法继续处理新的请求.

    除了自己定义 ThreadPoolExecutor 外. 还有其他方法. 如apache和guava等.

    四个构造函数

        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue)
                 
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory)
                 
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  RejectedExecutionHandler handler)
                 
        public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

    int corePoolSize => 该线程池中核心线程数最大值
    线程池新建线程的时候,如果当前线程总数小于corePoolSize, 则新建的是核心线程, 如果超过corePoolSize, 则新建的是非核心线程

    核心线程默认情况下会一直存活在线程池中, 即使这个核心线程啥也不干(闲置状态).

    如果指定 ThreadPoolExecutor 的 allowCoreThreadTimeOut 这个属性为 true, 那么核心线程如果不干活(闲置状态)的话, 超过一定时间(时长下面参数决定), 就会被销毁掉

    很好理解吧, 正常情况下你不干活我也养你, 因为我总有用到你的时候, 但有时候特殊情况(比如我自己都养不起了), 那你不干活我就要把你干掉了

    int maximumPoolSize
    该线程池中线程总数最大值

    线程总数 = 核心线程数 + 非核心线程数.

    long keepAliveTime
    该线程池中非核心线程闲置超时时长

    一个非核心线程, 如果不干活(闲置状态)的时长超过这个参数所设定的时长, 就会被销毁掉

    如果设置 allowCoreThreadTimeOut = true, 则会作用于核心线程

    TimeUnit unit

    keepAliveTime的单位, TimeUnit是一个枚举类型, 其包括:

    TimeUnit.DAYS;               //天
    TimeUnit.HOURS;             //小时
    TimeUnit.MINUTES;           //分钟
    TimeUnit.SECONDS;           //秒
    TimeUnit.MILLISECONDS;      //毫秒
    TimeUnit.MICROSECONDS;      //微妙
    TimeUnit.NANOSECONDS;       //纳秒

    BlockingQueue workQueue

    一个阻塞队列, 用来存储等待执行的任务. 也就是说现在有10个任务, 核心线程 有四个, 非核心线程有六个, 那么这六个线程会被添加到 workQueue 中, 等待执行.

    这个参数的选择也很重要, 会对线程池的运行过程产生重大影响, 一般来说, 这里的阻塞队列有以下几种选择:

    SynchronousQueue: 这个队列接收到任务的时候, 会直接提交给线程处理, 而不保留它, 如果所有线程都在工作怎么办? 那就*新建一个线程来处理这个任务!所以为了保证不出现<线程数达到了maximumPoolSize而不能新建线程>的错误, 使用这个类型队列的时候, maximumPoolSize 一般指定成 Integer.MAX_VALUE, 即无限大.

    LinkedBlockingQueue: 这个队列接收到任务的时候, 如果当前线程数小于核心线程数, 则核心线程处理任务; 如果当前线程数等于核心线程数, 则进入队列等待. 由于这个队列最大值为 Integer.MAX_VALUE , 即所有超过核心线程数的任务都将被添加到队列中,这也就导致了 maximumPoolSize 的设定失效, 因为总线程数永远不会超过 corePoolSize.

    ArrayBlockingQueue: 可以限定队列的长度, 接收到任务的时候, 如果没有达到 corePoolSize 的值, 则核心线程执行任务, 如果达到了, 则入队等候, 如果队列已满, 则新建线程(非核心线程)执行任务, 又如果总线程数到了maximumPoolSize, 并且队列也满了, 则发生错误.

    DelayQueue: 队列内元素必须实现 Delayed 接口, 这就意味着你传进去的任务必须先实现Delayed接口. 这个队列接收到任务时, 首先先入队, 只有达到了指定的延时时间, 才会执行任务.

    ThreadFactory threadFactory

    它是ThreadFactory类型的变量, 用来创建新线程.

    默认使用 Executors.defaultThreadFactory() 来创建线程. 使用默认的 ThreadFactory 来创建线程时, 会使新创建的线程具有相同的 NORM_PRIORITY 优先级并且是非守护线程, 同时也设置了线程的名称.

    RejectedExecutionHandler handler

    表示当拒绝处理任务时的策略, 有以下四种取值:

    ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常(默认).
    ThreadPoolExecutor.DiscardPolicy:直接丢弃任务, 但是不抛出异常.
    ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务, 然后重新尝试执行任务(重复此过程)
    ThreadPoolExecutor.CallerRunsPolicy:用调用者所在的线程来执行任务.

    【相关推荐:Java视频教程

    以上就是Java中ThreadPoolExecutor线程池的用法介绍的详细内容,更多请关注php中文网其它相关文章!

    声明:本文转载于:segmentfault,如有侵犯,请联系admin@php.cn删除
    专题推荐:java
    上一篇:Java多线程批量数据导入的方法介绍 下一篇:Java8的CompletableFuture的用法介绍(附示例)
    Web大前端开发直播班

    相关文章推荐

    • JavaScript作用域的全面解析(附代码)• Java开发中必须了解的基本数据类型你知道多少?• java基础反射Reflection简单介绍• Java多线程批量数据导入的方法介绍

    全部评论我要评论

  • 取消发布评论发送
  • 1/1

    PHP中文网