【并发编程篇】Java并发线程池底层原理详解与源码分析

public static final int MIN_PRIORITY = 1;
public static final int NORM_PRIORITY = 5;
public static final int MAX_PRIORITY = 10;

public static enum State {
    NEW,                           //新建
    RUNNABLE,              //就绪、运行
    BLOCKED,               //阻塞
    WAITING,               //等待
    TIMED_WAITING,   //计时等待
    TERMINATED;            //结束
}

public class CreateDemo {
    public static final int MAX_TURN = 5;
    public static String getCurThreadName() {
        return Thread.currentThread().getName();
    }
    //线程的编号
    static int threadNo = 1;

    static class DemoThread extends Thread {  //① 
        public DemoThread() {
            super("DemoThread-" + threadNo++); //②
        }

        public void run() {   //③
            for (int i = 1; i < MAX_TURN; i++) {
                Print.cfo(getName() + ", 轮次：" + i);
            }
            Print.cfo(getName() + " 运行结束.");
        }
    }

    public static void main(String args[]) throws InterruptedException {
        Thread thread = null;
        //方法一：使用Thread子类创建和启动线程
        for (int i = 0; i < 2; i++) {
            thread = new DemoThread();
            thread.start();
        }

        Print.cfo(getCurThreadName() + " 运行结束.");
    }
}

package java.lang;
@FunctionalInterface
    public interface Runnable {
        void run();
    }

public class CreateDemo2
    {
        public static final int MAX_TURN = 5;
        static int threadNo = 1;
        static class RunTarget implements Runnable  //①实现Runnable接口
            {
                public void run()  //②在这里编写业务逻辑
                {
                    for (int j = 1; j < MAX_TURN; j++)
                        {
                            Print.cfo(ThreadUtil.getCurThreadName() + ", 轮次：" + j);
                        }
                    Print.cfo(getCurThreadName() + " 运行结束.");
                }
            }

        public static void main(String args[]) throws InterruptedException
        {
            Thread thread = null;
            for (int i = 0; i < 2; i++)
                {
                    Runnable target = new RunTarget();
                    //通过Thread 类创建线程对象，将Runnable实例作为实际参数传入
                    thread = new Thread(target, "RunnableThread" + threadNo++);
                    thread.start();
                }
        }
    }

package java.util.concurrent;
@FunctionalInterface
    public interface Callable<V> {
        V call() throws Exception;
    }

package java.util.concurrent;
     
     public interface RunnableFuture<V>  extends  Runnable, Future<V> {
         void run();
     }

    package java.util.concurrent;
     public interface Future<V> {
         boolean cancel(boolean mayInterruptRunning); //取消异步执行
         boolean isCancelled();
         boolean isDone();//判断异步任务是否执行完成
         //获取异步任务完成后的执行结果
         V get() throws InterruptedException, ExecutionException;
         //设置时限，获取异步任务完成后的执行结果
         V get(long timeout, TimeUnit unit) throws InterruptedException, 
                                               ExecutionException, TimeoutException;
        ...
     }

public class CreateDemo3 {
         public static final int MAX_TURN = 5;
         public static final int COMPUTE_TIMES = 100000000;
     
     //①创建一个 Callable 接口的实现类
         static class ReturnableTask implements Callable<Long> {
              //②编写好异步执行的具体逻辑，可以有返回值
              public Long call() throws Exception{
                 long startTime = System.currentTimeMillis();
                 Print.cfo(getCurThreadName() + " 线程运行开始.");
                 Thread.sleep(1000);
     
                 for (int i = 0; i < COMPUTE_TIMES; i++) {
                     int j = i * 10000;
                 }
                 long used = System.currentTimeMillis() - startTime;
                 Print.cfo(getCurThreadName() + " 线程运行结束.");
                 return used;
             }
         }
     
         public static void main(String args[]) throws InterruptedException {
             ReturnableTask task=new ReturnableTask();//③
             FutureTask<Long> futureTask = new FutureTask<Long>(task);//④
             Thread thread = new Thread(futureTask, "returnableThread");//⑤
             thread.start();//⑥
             Thread.sleep(500);
             Print.cfo(getCurThreadName() + " 让子弹飞一会儿.");
             Print.cfo(getCurThreadName() + " 做一点自己的事情.");
             for (int i = 0; i < COMPUTE_TIMES / 2; i++) {
                 int j = i * 10000;
             }
     
             Print.cfo(getCurThreadName() + " 获取并发任务的执行结果.");
             try {
                 Print.cfo(thread.getName()+"线程占用时间："
                                          + futureTask.get());//⑦
             } catch (InterruptedException e) {
                 e.printStackTrace();
             } catch (ExecutionException e) {
                 e.printStackTrace();
             }
             Print.cfo(getCurThreadName() + " 运行结束.");
         }
     }

//创建一个包含三个线程的线程池
private static ExecutorService pool = Executors.newFixedThreadPool(3);

//方法一：执行一个 Runnable类型的target执行目标实例，无返回
void execute(Runnable command);

//方法二：提交一个 Callable类型的target执行目标实例, 返回一个Future异步任务实例
<T> Future<T> submit(Callable<T> task);  

//方法三：提交一个 Runnable类型的target执行目标实例, 返回一个Future异步任务实例
Future<?> submit(Runnable task);

public class CreateDemo4
    {

        public static final int MAX_TURN = 5;
        public static final int COMPUTE_TIMES = 100000000;

        //创建一个包含三个线程的线程池
        private static ExecutorService pool = Executors.newFixedThreadPool(3);

        static class DemoThread implements Runnable
            {
                @Override
                public void run()
                {
                    for (int j = 1; j < MAX_TURN; j++)
                        {
                            Print.cfo(getCurThreadName() + ", 轮次：" + j);
                            sleepMilliSeconds(10);
                        }
                }
            }


        static class ReturnableTask implements Callable<Long>
            {
                //返回并发执行的时间
                public Long call() throws Exception
                {
                    long startTime = System.currentTimeMillis();
                    Print.cfo(getCurThreadName() + " 线程运行开始.");
                    for (int j = 1; j < MAX_TURN; j++)
                        {
                            Print.cfo(getCurThreadName() + ", 轮次：" + j);
                            sleepMilliSeconds(10);
                        }
                    long used = System.currentTimeMillis() - startTime;
                    Print.cfo(getCurThreadName() + " 线程运行结束.");
                    return used;
                }
            }

        public static void main(String[] args) {

            pool.execute(new DemoThread()); //执行线程实例，无返回
            pool.execute(new Runnable()
                         {
                             @Override
                             public void run()
                             {
                                 for (int j = 1; j < MAX_TURN; j++)
                                     {
                                         Print.cfo(getCurThreadName() + ", 轮次：" + j);
                                         sleepMilliSeconds(10);
                                     }
                             }
                         });
            //提交Callable 执行目标实例，有返回
            Future future = pool.submit(new ReturnableTask());
            Long result = (Long) future.get();
            Print.cfo("异步任务的执行结果为：" + result);
            sleepSeconds(Integer.MAX_VALUE);
        }
    }

void execute(Runnable command)

 //向线程池提交单个异步任务
     <T> Future<T> submit(Callable<T> task);
     //向线程池提交批量异步任务
     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
             throws InterruptedException;

AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。AbstractExecutorService存在的目的是为ExecutorService中的接口提供默认实现。

ThreadPoolExecutor就是大名鼎鼎的“线程池”实现类，它继承于AbstractExecutorService抽象类。

ThreadPoolExecutor是JUC线程池的核心实现类。线程的创建和终止需要很大的开销，线程池中预先提供了指定数量的可重用线程，所以使用线程池会节省系统资源，并且每个线程池都维护了一些基础的数据统计，方便线程的管理和监控。

public class CreateThreadPoolDemo
    {
        public static final int SLEEP_GAP = 500;
        //异步任务的执行目标类
        static class TargetTask implements Runnable
            {
                static AtomicInteger taskNo = new AtomicInteger(1);
                private String taskName;
                public TargetTask()
                {
                    taskName = "task-" + taskNo.get();
                    taskNo.incrementAndGet();
                }
                public void run()
                {
                    Print.tco("任务：" + taskName + " doing");
                    // 线程睡眠一会
                    sleepMilliSeconds(SLEEP_GAP);
                    Print.tco(taskName + " 运行结束.");
                }
            }

        //测试用例：只有一个线程的线程池
        @Test
        public void testSingleThreadExecutor()
        {
            ExecutorService pool = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 5; i++)
                {
                    pool.execute(new TargetTask());
                    pool.submit(new TargetTask());
                }
            sleepSeconds(1000);
            //关闭线程池
            pool.shutdown();
        }
    }

public class CreateThreadPoolDemo
     {
         public static final int SLEEP_GAP = 500;
         //异步任务的执行目标类
         static class TargetTask implements Runnable
         {
         //为了节约篇幅，省略重复内容
         }
        //测试用例：只有3个线程固定大小的线程池
         @Test
         public void testNewFixedThreadPool()
         {
             ExecutorService pool = Executors.newFixedThreadPool(3);
             for (int i = 0; i < 5; i++)
             {
                 pool.execute(new TargetTask());
                 pool.submit(new TargetTask());
             }
             sleepSeconds(1000);
             //关闭线程池
             pool.shutdown();
         }
         // 省略其他
     }

public class CreateThreadPoolDemo
     {
         public static final int SLEEP_GAP = 500;
         //异步任务的执行目标类
         static class TargetTask implements Runnable
         {
         //为了节约篇幅，省略重复内容
         }
            //测试用例：“可缓存线程池”
             @Test
             public void testNewCacheThreadPool()
             {
                 ExecutorService pool = Executors.newCachedThreadPool();
                 for (int i = 0; i < 5; i++)
                 {
                     pool.execute(new TargetTask());
                     pool.submit(new TargetTask());
                 }
                 sleepSeconds(1000);
                 //关闭线程池
                 pool.shutdown();
             }
         // 省略其他
     }

//方法一：创建一个可调度线程池，池内仅含有一个线程
     public static ScheduledExecutorService newSingleThreadScheduledExecutor();
     
     //方法二：创建一个可调度线程池，池内含有N个线程，N的值为输入参数corePoolSize
     public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) ;

public class CreateThreadPoolDemo
     {
         public static final int SLEEP_GAP = 500;
         //异步任务的执行目标类
         static class TargetTask implements Runnable
         {
         //为了节约篇幅，省略重复内容
         }
     
         //测试用例：“可调度线程池”
         @Test
         public void testNewScheduledThreadPool()
         {
              ScheduledExecutorService scheduled = 
                                Executors.newScheduledThreadPool(2);
              for (int i = 0; i < 2; i++)
             {
                 scheduled.scheduleAtFixedRate(new TargetTask(),
                         0, 500, TimeUnit.MILLISECONDS);
                 // 以上参数中：0表示首次执行任务的延迟时间，500表示每次执行任务的间隔时间
                 // TimeUnit.MILLISECONDS是执行的时间间隔数值，单位为毫秒
             }
             sleepSeconds(1000);
             //关闭线程池
             scheduled.shutdown();
         }
     // 省略其他
     }