Java并发编程一四种常用线程池简述

雨点打透心脏的1/2处 2023-07-05 14:52 60阅读 0赞

推荐：[Java并发编程汇总][Java]

## Java并发编程一四种常用线程池简述 ##

我们使用线程是用来执行任务的，所以，首先我们得有一个任务，这里我们定义线程的任务很简单，就是输出自己（当前线程）的`name`，为了方便观察，在每个线程的执行逻辑中`sleep(500)`。

**任务**

`Task`类，实现了`Runnable`接口。

package threadpool;
    
    class Task implements Runnable { 
    
        @Override
        public void run() { 
            try { 
                Thread.sleep(500);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName());
        }
    }

**SingleThreadExecutor**

测试代码：

package threadpool;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class SingleThreadExecutor { 
        public static void main(String[] args) { 
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 1000; i++) { 
                executorService.execute(new Task());
            }
        }
    }

`ExecutorService`接口继承了`Executor`接口，它比`Executor`接口有更丰富的方法，便于我们使用。  
`Executors`是Java线程池的工具类。

输出：

pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-1

上面的输出其实只是一部分，很显然会输出一千个`pool-1-thread-1`。

`SingleThreadExecutor`这种线程池只会创建一个线程去执行所有的任务。

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

`corePoolSize=1`，`maximumPoolSize=1`，很明显核心线程数和最大线程数都是`1`，这里的`keepAliveTime=0`说明线程池不需要去试图销毁线程，因为`keepAliveTime`对核心线程是没有约束的，又`corePoolSize`和`maximumPoolSize`都是`1`，即线程池的线程都是核心线程。

`workQueue`是`LinkedBlockingQueue`，等以后写阻塞队列相关的博客时，再详细介绍`LinkedBlockingQueue`，现在我们只需要知道`LinkedBlockingQueue`这种阻塞队列，调用它的无参构造器，创建的阻塞队列的容量是`Integer.MAX_VALUE`，你可以认为是无穷大。这也很符合`SingleThreadExecutor`的特性，这种线程池只会创建一个线程来执行任务，而线程池创建`corePoolSize`个核心线程后，需要等待任务队列满了之后再去创建其他的线程来帮忙，而`SingleThreadExecutor`这种线程池的任务队列容量无穷大（这里认为`Integer.MAX_VALUE`为无穷大），也就无需创建核心线程以外的线程来帮忙了。

/** * Creates a {@code LinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}. */
        public LinkedBlockingQueue() { 
            this(Integer.MAX_VALUE);
        }

**FixedThreadPool**

测试代码：

package threadpool;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class FixedThreadPoolTest { 
    
        public static void main(String[] args) { 
            ExecutorService executorService = Executors.newFixedThreadPool(4);
            for (int i = 0; i < 1000; i++) { 
                executorService.execute(new Task());
            }
        }
    }

输出：

pool-1-thread-2
    pool-1-thread-3
    pool-1-thread-4
    pool-1-thread-1
    pool-1-thread-4
    pool-1-thread-2
    pool-1-thread-1
    pool-1-thread-3
    pool-1-thread-3
    pool-1-thread-2
    pool-1-thread-1
    pool-1-thread-4
    pool-1-thread-2
    pool-1-thread-4
    pool-1-thread-3

上面的输出也是输出的一部分，这种线程池可以和`SingleThreadExecutor`进行对比理解，`SingleThreadExecutor`创建一个线程执行任务，而`FixedThreadPool`创建固定数量的线程执行任务。

/** * Creates a thread pool that reuses a fixed number of threads * operating off a shared unbounded queue. 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 * @return the newly created thread pool * @throws IllegalArgumentException if {@code nThreads <= 0} */
        public static ExecutorService newFixedThreadPool(int nThreads) { 
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }

和`SingleThreadExecutor`的创建方法很相似，就不多说了。

**CachedThreadPool**

测试代码：

package threadpool;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class CachedThreadPool { 
    
        public static void main(String[] args) { 
            ExecutorService executorService = Executors.newCachedThreadPool();
            for (int i = 0; i < 1000; i++) { 
                executorService.execute(new Task());
            }
        }
    }

输出：

pool-1-thread-791
    pool-1-thread-868
    pool-1-thread-864
    pool-1-thread-790
    pool-1-thread-853
    pool-1-thread-789
    pool-1-thread-863
    pool-1-thread-717
    pool-1-thread-776
    pool-1-thread-865
    pool-1-thread-861
    pool-1-thread-782
    pool-1-thread-857
    pool-1-thread-859
    pool-1-thread-858
    pool-1-thread-751
    pool-1-thread-692

上面的输出也是输出的一部分，可见这种线程池创建线程就比较随意了。

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

`SynchronousQueue`实际上不是一个真正的队列，因为它不会为队列中元素维护存储空间，与其他队列不同的是，它维护一组线程，这些线程在等待着把元素（任务）加入或移出队列，所以可以知道，当任务来了，`SynchronousQueue`就把任务交给线程池，如果线程池没有空闲的线程，就会创建线程去处理这个任务，这就是为什么这种线程池创建线程看起来比较随意了，`corePoolSize=0`说明这种线程池没有核心线程，也就是线程池可以试图去销毁所有闲置的线程，`maximumPoolSize=Integer.MAX_VALUE`可以认为这种线程池可以无限地创建线程（这里认为`Integer.MAX_VALUE`为无穷大）。

图示如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]  
这里的`keepAliveTime=60L`说明线程池会试图销毁线程，当线程闲置60秒（单位是秒）后就可能会被销毁。

**ScheduledThreadPool**

测试代码：

package threadpool;
    
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    
    public class ScheduledThreadPoolTest { 
    
        public static void main(String[] args) { 
            ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(10);
            threadPool.schedule(new Task(), 5, TimeUnit.SECONDS);
        }
    }

输出：

pool-1-thread-1

/** * Creates and executes a one-shot action that becomes enabled * after the given delay. * * @param command the task to execute * @param delay the time from now to delay execution * @param unit the time unit of the delay parameter * @return a ScheduledFuture representing pending completion of * the task and whose {@code get()} method will return * {@code null} upon completion * @throws RejectedExecutionException if the task cannot be * scheduled for execution * @throws NullPointerException if command is null */
        public ScheduledFuture<?> schedule(Runnable command,
                                           long delay, TimeUnit unit);

`delay`为`5`，单位为秒，也就是延迟`5`秒执行，并且这个调用只会执行一次任务。

测试代码：

输出：

pool-1-thread-1
    pool-1-thread-1
    pool-1-thread-2
    pool-1-thread-1
    pool-1-thread-3
    pool-1-thread-2
    pool-1-thread-4
    pool-1-thread-1
    pool-1-thread-5
    pool-1-thread-3

输出只是一部分。

/** * Creates and executes a periodic action that becomes enabled first * after the given initial delay, and subsequently with the given * period; that is executions will commence after * {@code initialDelay} then {@code initialDelay+period}, then * {@code initialDelay + 2 * period}, and so on. * If any execution of the task * encounters an exception, subsequent executions are suppressed. * Otherwise, the task will only terminate via cancellation or * termination of the executor. If any execution of this task * takes longer than its period, then subsequent executions * may start late, but will not concurrently execute. * * @param command the task to execute * @param initialDelay the time to delay first execution * @param period the period between successive executions * @param unit the time unit of the initialDelay and period parameters * @return a ScheduledFuture representing pending completion of * the task, and whose {@code get()} method will throw an * exception upon cancellation * @throws RejectedExecutionException if the task cannot be * scheduled for execution * @throws NullPointerException if command is null * @throws IllegalArgumentException if period less than or equal to zero */
        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                      long initialDelay,
                                                      long period,
                                                      TimeUnit unit);

`initialDelay=1`、`period=1`，单位为秒，也就是延迟`1`秒执行任务，并且以`1`秒为一个周期一直执行任务。

显然这种线程池，很适合执行具有周期性的任务。

看看这种线程池的创建方法：

/** * Creates a new {@code ScheduledThreadPoolExecutor} with the * given core pool size. * * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @throws IllegalArgumentException if {@code corePoolSize < 0} */
        public ScheduledThreadPoolExecutor(int corePoolSize) { 
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }

`maximumPoolSize=Integer.MAX_VALUE`、`keepAliveTime=0`，这些应该不需要解释了吧，通过类比也可以知道，我们现在先不要去管`DelayedWorkQueue`。

[Java]: https://kaven.blog.csdn.net/article/details/104233588
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200217172449382.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70