Java多线程-线程池

线程池定义

• 多线程的程序可以最大限度的发挥多核CPU的性能，但是如果不对线程进行合理的管控，会影响整体系统的性能。线程虽然相对进程来说很轻量级，但是也要占据一定的内存，新建和销毁也需要时间，不能无节制的创建线程。
• 线程池本质上是一组线程实时处理休眠状态，等待唤醒执行，使用线程池中，创建线程变成了从线程池中获取线程。
• 线程池可以较少创建和销毁线程上所花的时间以及系统资源的开销，同时将当前任务和主线程任务隔离，实现和主线程的异步执行，提高程序的执行效率。

线程池的种类

JDK对线程池的支持如下:

(图片来源：http://www.blogjava.net/xylz/archive/2010/12/21/341281.html)

Java中提供了四个静态方法来创建一个线程池，在线程池中执行任务比为每个任务分配一个线程优势更多，通过重用现有的线程而不是创建新线程，可以在处理多个请求时分摊线程的创建和销毁的开销。当请求到达时，线程已经存在，提高了响应速度。通过配置线程池的大小，可以创建爱你足够多的线程充分利用CPU的多核资源，同时防止线程太多耗尽资源。

newFixedThreadPool()

• 该方法返回一个固定线程数量的线程池。
• 该线程池中的线程数量始终不变。
• 当有一个新的任务提交时，线程池中若有空闲线程，则立即执行
• 如果没有空闲线程，则会将任务保存在一个任务队列中，线程空闲时，处理线程任务队列中的数据

Executors中方法

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

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

使用实例

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo {
    public static class MyTask implements Runnable {

        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ":Thread ID: " + Thread.currentThread().getId());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        MyTask task = new MyTask();
        ExecutorService service = Executors.newFixedThreadPool(5);
//        ExecutorService service = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            service.execute(task);
        }
    }
}
/*
1535943816251:Thread ID: 11
1535943816251:Thread ID: 12
1535943816251:Thread ID: 13
1535943816251:Thread ID: 14
1535943816251:Thread ID: 15
1535943817253:Thread ID: 11
1535943817253:Thread ID: 13
1535943817253:Thread ID: 12
1535943817253:Thread ID: 14
1535943817253:Thread ID: 15
*/

newSingleThreadExecutor()

• 该方法返回一个只有一个线程的线程池
• 当有一个新的任务提交时，空闲则执行，否则等待执行
• 确保线程之间的串行执行

Executors中方法

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

使用实例

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo {
    public static class MyTask implements Runnable {

        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ":Thread ID: " + Thread.currentThread().getId());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        MyTask task = new MyTask();
//        ExecutorService service = Executors.newFixedThreadPool(5);
        ExecutorService service = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            service.execute(task);
        }
    }
}
/*
1535944120085:Thread ID: 11
1535944120086:Thread ID: 12
1535944120086:Thread ID: 13
1535944120086:Thread ID: 14
1535944120087:Thread ID: 15
1535944120087:Thread ID: 16
1535944120088:Thread ID: 17
1535944120088:Thread ID: 18
1535944120088:Thread ID: 19
1535944120088:Thread ID: 20
*/

newCachedThreadPool()

• 返回一个可根据实际情况调整线程数量的线程池
• 线程池中线程的数量不确定
• 任务提交时，复用线程优于新建线程

Executors中方法

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

使用实例

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo {
    public static class MyTask implements Runnable {

        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + ":Thread ID: " + Thread.currentThread().getId());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        MyTask task = new MyTask();
//        ExecutorService service = Executors.newFixedThreadPool(5);
//        ExecutorService service = Executors.newCachedThreadPool();
        ExecutorService service = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            service.execute(task);
        }
    }
}
/*
1535944197414:Thread ID: 11
1535944198425:Thread ID: 11
1535944199425:Thread ID: 11
1535944200425:Thread ID: 11
1535944201426:Thread ID: 11
1535944202426:Thread ID: 11
1535944203426:Thread ID: 11
1535944204427:Thread ID: 11
1535944205427:Thread ID: 11
1535944206428:Thread ID: 11
*/

newScheduleThreadPool()

• 返回一个ScheduledExecutorService对象
• 增加了定时执行的功能，如延时执行或者周期执行

Executors中方法

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

使用实例

/**
 * 计划任务
 */
public class ScheduledExecutorServiceDemo {
    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
        ses.scheduleAtFixedRate((Runnable) () -> {
            try {
                Thread.sleep(2000);
                System.out.println(System.currentTimeMillis()/1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 0, 8, TimeUnit.SECONDS);
    }
}
/*
1535945140
1535945148
1535945156
1535945164
*/

四种定时执行的方式

/**
 * 延时执行
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Void>(command, null,
                                      triggerTime(delay, unit)));
    delayedExecute(t);
    return t;
}

/**
 * 带饭后之的延迟执行
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                       long delay,
                                       TimeUnit unit) {
    if (callable == null || unit == null)
        throw new NullPointerException();
    RunnableScheduledFuture<V> t = decorateTask(callable,
        new ScheduledFutureTask<V>(callable,
                                   triggerTime(delay, unit)));
    delayedExecute(t);
    return t;
}

/**
 * 按照指定频率周期执行某个任务
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 * @throws IllegalArgumentException   {@inheritDoc}
 */
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0)
        throw new IllegalArgumentException();
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

/**
 * 周期定时执行某个任务/按指定频率间隔执行魔鬼任务
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 * @throws IllegalArgumentException   {@inheritDoc}
 */
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (delay <= 0)
        throw new IllegalArgumentException();
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(-delay));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

submit

• submit()与execute()类似，区别在于一个有返回值一个没有
• submit()适合生产者消费者模式，与Future一起使用，可以在没有返回结果时阻塞当前线程的作用
• 实现源码:

/**
 * Submits a value-returning task for execution and returns a
 * Future representing the pending results of the task. The
 * Future's {@code get} method will return the task's result upon
 * successful completion.
 *
 * <p>
 * If you would like to immediately block waiting
 * for a task, you can use constructions of the form
 * {@code result = exec.submit(aCallable).get();}
 *
 * <p>Note: The {@link Executors} class includes a set of methods
 * that can convert some other common closure-like objects,
 * for example, {@link java.security.PrivilegedAction} to
 * {@link Callable} form so they can be submitted.
 *
 * @param task the task to submit
 * @param <T> the type of the task's result
 * @return a Future representing pending completion of the task
 * @throws RejectedExecutionException if the task cannot be
 *         scheduled for execution
 * @throws NullPointerException if the task is null
 */
<T> Future<T> submit(Callable<T> task);

/**
 * Submits a Runnable task for execution and returns a Future
 * representing that task. The Future's {@code get} method will
 * return the given result upon successful completion.
 *
 * @param task the task to submit
 * @param result the result to return
 * @param <T> the type of the result
 * @return a Future representing pending completion of the task
 * @throws RejectedExecutionException if the task cannot be
 *         scheduled for execution
 * @throws NullPointerException if the task is null
 */
<T> Future<T> submit(Runnable task, T result);

/**
 * Submits a Runnable task for execution and returns a Future
 * representing that task. The Future's {@code get} method will
 * return {@code null} upon <em>successful</em> completion.
 *
 * @param task the task to submit
 * @return a Future representing pending completion of the task
 * @throws RejectedExecutionException if the task cannot be
 *         scheduled for execution
 * @throws NullPointerException if the task is null
 */
Future<?> submit(Runnable task);

ThreadPooLExecutor类

• 线程池中最重要的类

基本参数

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 线程池状态
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// 自动装箱和拆箱
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 线程池状态显示
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}

private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}

private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

// 线程池缓冲队列：当线程池线程数量超过corePoolSize时，待运行的线程会放入到这个队列中
private final BlockingQueue<Runnable> workQueue;

// 重入锁，更新核心线程池代销，最大线程池大小时进行加锁
private final ReentrantLock mainLock = new ReentrantLock();

// 运行线程集合
private final HashSet<Worker> workers = new HashSet<Worker>();

// awaitTermination支持
private final Condition termination = mainLock.newCondition();

// 用来记录线程池中曾经出现过的最大线程数
private int largestPoolSize;

// 完成的线程数量
private long completedTaskCount;

// 线程工厂
private volatile ThreadFactory threadFactory;

// 拒绝Handler
private volatile RejectedExecutionHandler handler;

// 线程执行完毕后在线程池中的缓存时间，超过这个时长，线程会被回收
private volatile long keepAliveTime;

// 是否开启keepAliveTime
private volatile boolean allowCoreThreadTimeOut;

// 核心线程池大小
private volatile int corePoolSize;

// 最大线程池大小
private volatile int maximumPoolSize;

构造方法

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

• 缓存队列的类型

  • ArrayBlockingQueue : 有界的数组队列
  • LinkedBlockingQueue : 可支持有界/无界的队列，使用链表实现
  • PriorityBlockingQueue : 优先队列，可以针对任务排序
  • SynchronousQueue : 队列长度为1的队列，和Array有点区别就是：client thread提交到block queue会是一个阻塞过程，直到有一个worker thread连接上来poll task。当线

• 拒绝策略

  • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
  • ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
  • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
  • ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

execute

• 表示向线程池中添加线程，可能会立即执行，也可能不会。
• 无法预知线程的开始和结束

/**
 * Executes the given task sometime in the future.  The task
 * may execute in a new thread or in an existing pooled thread.
 *
 * If the task cannot be submitted for execution, either because this
 * executor has been shutdown or because its capacity has been reached,
 * the task is handled by the current {@code RejectedExecutionHandler}.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 *         {@code RejectedExecutionHandler}, if the task
 *         cannot be accepted for execution
 * @throws NullPointerException if {@code command} is null
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    // 如果运行线程小于corePoolSize，则创建一个新的线程
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 如果运行的线程等于corePoolSize，则进入workQueue等待队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 不能进入等待队列，尝试创建一个新的线程，失败则执行拒绝任务
    else if (!addWorker(command, false))
        reject(command);
}

• ThreadPoolExecutor中，包含了一个任务缓存队列和若干个执行线程，任务缓存队列是一个大小固定的缓冲区队列，用来缓存待执行的任务，执行线程用来处理待执行的任务。每个待执行的任务，都必须实现Runnable接口，执行线程调用其run()方法，完成相应任务。
• ThreadPoolExecutor对象初始化时，不创建任何执行线程，当有新任务进来时，才会创建执行线程。
• 构造ThreadPoolExecutor对象时，需要配置该对象的核心线程池大小和最大线程池大小：
• 当目前执行线程的总数小于核心线程大小时，所有新加入的任务，都在新线程中处理
• 当目前执行线程的总数大于或等于核心线程时，所有新加入的任务，都放入任务缓存队列中
• 当目前执行线程的总数大于或等于核心线程，并且缓存队列已满，同时此时线程总数小于线程池的最大大小，那么创建新线程，加入线程池中，协助处理新的任务。
• 当所有线程都在执行，线程池大小已经达到上限，并且缓存队列已满时，就rejectHandler拒绝新的任务

shutdown

• 放在execute后面
• 表示当前线程池不接受新添加的线程
• 所有线程执行完毕时，回收线程池资源
• 不会马上关闭线程池，执行完已经提交的线程池

/**
 * Initiates an orderly shutdown in which previously submitted
 * tasks are executed, but no new tasks will be accepted.
 * Invocation has no additional effect if already shut down.
 *
 * <p>This method does not wait for previously submitted tasks to
 * complete execution.  Use {@link #awaitTermination awaitTermination}
 * to do that.
 *
 * @throws SecurityException {@inheritDoc}
 */
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

shutdownNow

• 立刻关闭线程池，会引起系统数据异常

/**
 * Attempts to stop all actively executing tasks, halts the
 * processing of waiting tasks, and returns a list of the tasks
 * that were awaiting execution. These tasks are drained (removed)
 * from the task queue upon return from this method.
 *
 * <p>This method does not wait for actively executing tasks to
 * terminate.  Use {@link #awaitTermination awaitTermination} to
 * do that.
 *
 * <p>There are no guarantees beyond best-effort attempts to stop
 * processing actively executing tasks.  This implementation
 * cancels tasks via {@link Thread#interrupt}, so any task that
 * fails to respond to interrupts may never terminate.
 *
 * @throws SecurityException {@inheritDoc}
 */
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

实例

import java.io.Serializable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @program: 03_JDK_Concurence_Package
 * @author: cdx
 * @create: 2018-09-03 12:34
 **/
public class ThreadPoolTaskDemo {
    public static final int PRODUCT_TASK_SLEEP_TIME = 2;
    public static final int PRODUCT_TASK_MAX_NUMBER = 10;

    public static void main(String[] args) {
        // 创建一个线程池，核心线程池大小为2，最大线程池大小为4，线程缓存3秒钟，等待队列数组，拒绝侧率直接抛弃
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2,
                4,
                3,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());

        for (int i = 0; i <= PRODUCT_TASK_MAX_NUMBER; i++) {
            try {
                String task = "task@ " + i;
                System.out.println("put " + task);
                executor.execute(new ThreadPooLTask(task));
                Thread.sleep(PRODUCT_TASK_SLEEP_TIME);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

class ThreadPooLTask implements Runnable, Serializable {
    private Object threadPoolTskData;

    public ThreadPooLTask(Object threadPoolTskData) {
        this.threadPoolTskData = threadPoolTskData;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "start .." + threadPoolTskData);
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        threadPoolTskData = null;
    }

    public Object getTask() {
        return this.threadPoolTskData;
    }
}
/*
put task@ 0
pool-1-thread-1start ..task@ 0
put task@ 1
pool-1-thread-2start ..task@ 1
put task@ 2
put task@ 3
put task@ 4
put task@ 5
pool-1-thread-3start ..task@ 5
put task@ 6
pool-1-thread-4start ..task@ 6
put task@ 7
put task@ 8
put task@ 9
put task@ 10
pool-1-thread-1start ..task@ 8
pool-1-thread-2start ..task@ 9
pool-1-thread-3start ..task@ 10
 */