java 线程池

移动开发 waitig 483℃ 百度已收录 0评论

一 什么是线程池

顾名思义,一个池子里面用来存放一定数量的线程资源。类似的概念还有socket连接池(okhttp)、数据库连接池。池子里面包含了几个概念,1.一个容器用来存放指定数量的资源;2.池子里面的资源可以复用。

二 为什么用线程池

多线程可以提升多核cpu的利用率。线程少,cpu时间片切换,线程上下文切换开销比较大;线程太多会增加内存开销,并且线程创建、销毁的代价也比较高。权衡cpu利用率提升与线程创建开销,创建与cpu核心数相关数量的线程资源,并且在线程工作完之后回收以便后面复用,这样就引出了线程池的概念。

线程池的作用:避免频繁地创建和销毁线程,达到线程对象的重用。另外,使用线程池还可以根据项目灵活地控制并发的数目。

三 怎么设计线程池

以java线程池为例,核心类是java/util/concurrent/ThreadPoolExecutor.java,有四个构造方法,如下:

 // Public constructors and methods

    /**
     * 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;
    }

总共七个参数:

  1. corePoolSize
  2. maximumPoolSize
  3. workQueue
  4. unit
  5. keepAliveTime
  6. threadFactory
  7. handler

corePoolSize,maximumPoolSize用来定义线程池中线程的数量大小。使用规则如下:

  1. 如果当前线程池中线程的数目低于corePoolSize,则创建新线程执行提交的任务,而无需检查当前是否有空闲线程

  2. 如果提交任务时线程池中线程数已达到corePoolSize,则将提交的任务加入等待执行队列

  3. 如果提交任务时等待执行的任务队列是有限队列,队列已满,且线程池中的线程数量

   /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

<1> corePoolSize为0,maximumPoolSize为无限大。理论上可以有任务提交过来就可以创建新线程来处理。

<2> 如果一个线程空闲超过60s将会被从线程池里清除

<3> 消息队列使用SynchronousQueue,同步队列,消息不会缓存到消息队列,而是会等待线程池内分配线程来直接处理

<4> 当一个线程处理完任务会被回收到线程池统一管理,下次被重用。

<5> 只适合用来处理短时间消耗的任务

  1. newFixedThreadPool
 /**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue, using the provided
     * ThreadFactory to create new threads when needed.  At any point,
     * at most {@code nThreads} threads will be active processing
     * tasks.  If additional tasks are submitted when all threads are
     * active, they will wait in the queue until a thread is
     * available.  If any thread terminates due to a failure during
     * execution prior to shutdown, a new one will take its place if
     * needed to execute subsequent tasks.  The threads in the pool will
     * exist until it is explicitly {@link ExecutorService#shutdown
     * shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     * @throws NullPointerException if threadFactory is null
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

<1> 使用固定大小数量的线程池,控制并发数,避免资源耗尽。

<2> 如果线程池中的线程数量没有达到maximumPoolSize,会新建一个线程来处理消息任务,使用完回收到线程池管理留后面重用。

<3> 如果线程池中的线程数量等于maximumPoolSize,新来的任务会在LinkedBlockingQueue这个无界队列中等待。

<4> 适合任务之间无依赖关系的场景

  1. newSingleThreadExecutor
   /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * {@code newFixedThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

可以理解为数量设置为1的newFixedThreadPool。适合用于数据库操作,文件操作,应用批量安装,应用批量删除等不适合并发但可能IO阻塞性及影响UI线程响应的操作。

  1. ScheduledExecutorService
  /**
     * Creates a thread pool that can schedule commands to run after a
     * given delay, or to execute periodically.
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle
     * @return a newly created scheduled thread pool
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

增加了定时执行的功能。

<1> 一定时间延时后开始执行任务。

<2> 周期性的执行任务


本文由【waitig】发表在等英博客
本文固定链接:java 线程池
欢迎关注本站官方公众号,每日都有干货分享!
等英博客官方公众号
点赞 (0)分享 (0)