（六）Java多线程——线程池

青旅半醒 2022-10-11 12:10 316阅读 0赞

> 线程池首先有几个接口先了解第一个是Executor，第二个是ExecutorService  
> 再后面才是线程池的一个使用ThreadPoolExecutor

### 1、Executor ###

Executor 顾名思义执行者，所有他有一个方法叫执行，执行的东西叫Runnnable,由于它是一个接口，所以它有很多实现，因此我们说有了Executor之后呢，我们线程就是一个任务的定义，比如RUnnable起了一个命令的意思，他的定义和运行就可以分开了，不像我们以前定义一个Thread，new一个Thread然后去重写的Run方法.start才可以运行，或者以前就是你写了一个Runnable你也必须的new一个Thread出来，一起的这种定义和运行是固定的，是写死的就是你new一个Thread让他出来运行。不用你亲自去指定每一个Thread，他的运行方式你可以自己去定义了，所以至于是怎么去定义的就看你怎么实现Executor接口了。这里是定义和运行分开这个一个含义，所以这个接口提现的是这个意思。所以这个接口就比较简单，至于你是直接调用Run还是new一个Thread那是你自己的事。

### 2、ExecutorService ###

他是从Executor继承，另外，他除了去实现Executor可以去执行一个任务之外，他还完善了整个任务执行器的一个生命周期，就拿线程池来举例子，一个线程池里面一堆的线程就是一堆的工人，执行完一个任务之后我这个线程怎么结束啊，线程池定义了这些一些方法：

void shutdown();//结束
    List<Runnable>shutdownNow();//马上结束
    boolean isShutdown();//是否结束了
    boolean isTerminated();//是不是整体都执行完了
    boolean awaitTermination(long timeout,TimeUnit unit)
    	throws InterruptedException;//等待技术，等多长时间，时间到了还不结束的话他就返回false

等等，所以这里面他是实现了一些个线程的线程池的生命周期的东西，扩展了Executor的接口，真正的线程池的实现是ExecutorSerivce执行任务之外还有submit提交任务，执行任务是直接拿过来马上运行，而submit是扔给这个线程池，什么时候运行由这个线程池来决定，相当于是异步的，我只要往里面一扔就不管了。那好，如果不管的话什么时候他有结果啊，这尼玛就涉及了比较新的类：比如说Future,RunnableFuture,FutureTask 。

以前定义一个线程的任务只能去实现Runnable接口，那在1.5之后他就增加Callable这个接口。  
Callable是什么，他类似于Runnable，不过Callable可以有返回值。  
有了Callable之后呢Future代表的是哪个Callable被执行完了之后，他会把结果封装到一个Futrue里面。  
由于Callable基本上就是为了线程池而设计的。Callable一般是配合线程池和Future来用的。  
更灵活的一个用法是FutureTask,既是一个Future同时又是一个Task。相当于是我自己可以作用一个任务来用，同时这个任务完成之后的结果页存在于这个对象里。

public class TestFutureTask { 
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            FutureTask<Integer> task = new FutureTask<>(()->{ 
                TimeUnit.MILLISECONDS.sleep(1000);
                return 100;
            });
    
            new Thread(task).start();
    
            System.out.println(task.get());//阻塞
        }
    }

CompleableFutue，管理多个Future的结果。  
例子：假设你能够提供一个服务，这个服务可以查询各大电商网站同一商品的价格并汇总展示

public class TestCompletableFuture { 
    
        /** * 假设你能够提供一个服务 * 这个服务可以查询各大电商网站同一商品的价格并汇总展示 */
        public static void main(String[] args) { 
            long start, end;
    
            start = System.currentTimeMillis();
    
            CompletableFuture<Double> futureTB = CompletableFuture.supplyAsync(() -> priceOfTB());//淘宝
            CompletableFuture<Double> futureTM = CompletableFuture.supplyAsync(() -> priceOfTM());//天猫
            CompletableFuture<Double> futureJD = CompletableFuture.supplyAsync(() -> priceOfJD());//京东
    
            CompletableFuture.allOf(futureTB,futureTM,futureJD).join();
    
            CompletableFuture.supplyAsync(() -> priceOfTB())
                    .thenApply(String::valueOf)
                    .thenApply(str -> "price " + str)
                    .thenAccept(System.out::println);
    
            end = System.currentTimeMillis();
            System.out.println("use completable future!" + (end - start));
    
            try { 
                System.in.read();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        private static void delay() { 
            int time = new Random().nextInt(500);
            try { 
                TimeUnit.MILLISECONDS.sleep(time);
                System.out.printf("After %s sleep!\n", time);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        private static double priceOfTB(){ 
            delay();
            return 1.00;
        }
        private static double priceOfTM(){ 
            delay();
            return 2.00;
        }
        private static double priceOfJD(){ 
            delay();
            return 3.00;
        }
    
        private static double priceOfAmazon(){ 
            delay();
            throw new RuntimeException("product net exist!");
        }
    }

### 3、ThreadPoolExecutor 线程池 ###

线程池从目前JDK提供的有两种类型，第一个就是普通的线程池ThreadPoolExecutor,  
第二种是ForkJoinPool,这两种是不同类型的线程池，能干的事不太一样。Fork分叉，分叉完再分叉，最后的结果汇总这叫join。

ThreadPoolExecutor 他的父类是从AbstractExecutorService,而AbstractExecutorService的父类是执行器，就是大家可以向这个池子里扔任务，让这个线程池去运行。

线程池定义：  
这里面的七个参数：

*  第一个**corePoolSize核心线程数**，最开始的时候这个线程池里面有一定的核心线程数的。
 *  第二个**maximumPoolSize最大线程数**，线程数不够了，能扩展到最大线程是多少。
 *  第三个**keepAliveTime生存时间**，意思是这个线程有很长时间没干活了请你把他归还给操作系统。
 *  第四个**TimeUnit.SECONDS生存时间的单位**，到底是毫秒，纳秒还是秒自己去定义。
 *  第五个**任务队列**就是之前讲到的BlockingQueue,各种各样的BlockingQueue你都可以往里面扔，我这用的是ArrayBlockingQueue,参数最多可以装四个任务；
 *  第六个**线程工厂**defaultThreadFactory，他返回的是一个new DefaultThreadFactory,他要你去实现ThreadFactory的接口，这个接口只有一个方法叫new Thread,所以就是产生线程的，可以通过这种方式产生自定义的线程池，默认产生的是defaultThreadFactory，而defaultThreadFactory产生线程的时候有几个特点：new出来的时候指定了group制定了线程名字，然后指定的这个线程绝对不是守护线程，设定好你线程的优先级。自己可以定义产生的到底是什么样的线程，指定线程名叫什么（作用是方便出错时候回溯）
 *  第七个**拒绝策略**，指线程池忙，而且任务队列满这种情况下我们就要执行各种各样的拒绝策略，jdk默认提供了四种拒绝策略，也是可以自己定义的。

1.  Abort：抛异常
2.  Discard：扔掉，不抛异常
3.  DiscardOldest：扔掉排队时间最久的
4.  CallerRuns：调用者处理服务  
    一般情况这四种我们会自定义策略，去实现这个拒绝策略的接口，处理的方式是一般我们得消息需要保存下来，要是订单的话那就更需要保存了，保存到kafka，保存到redis或者存到数据库随便你然后做好日志。

代码解释：
    定义了一个任务Task，这个任务是实现Runnable接口，就是一个普通的任务了，每一个任务里有一个编号i，
    然后打印这个编号，主要干这个事，打印完后阻塞System.in.read(),每个任务都是阻塞的，线程池维护
    这两个集合，第一个是线程的集合，里面是一个一个的线程。第二个是任务的集合，里面是一个一个的任务。
    这叫一个完整的线程池。

import java.io.IOException;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /** * @program: bike-lease * @description: 线程池定义 * @author: Jack.Fang * @date:2021-06-30 下午 3:07 **/
    public class TestThreadPoolExecutor { 
        static class Task implements Runnable { 
            private int i;
    
            public Task(int i) { 
                this.i = i;
            }
    
            @Override
            public void run() { 
                System.out.println(Thread.currentThread().getName() + " Task " + i);
                try { 
                    System.in.read();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
    
            @Override
            public String toString() { 
                return "Task{" +
                        "i=" + i +
                        '}';
            }
        }
    
        public static void main(String[] args) { 
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    2,
                    4,
                    60,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(4),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.CallerRunsPolicy());
    
            for(int i=0;i<8;i++){ 
                threadPoolExecutor.execute(new Task(i));
            }
            System.out.println(threadPoolExecutor.getQueue());
            threadPoolExecutor.execute(new Task(100));
            System.out.println(threadPoolExecutor.getQueue());
            threadPoolExecutor.shutdown();
        }
    
    }

结果：  
pool-1-thread-1 Task 0  
pool-1-thread-2 Task 1  
\[Task\{i=2\}, Task\{i=3\}, Task\{i=4\}, Task\{i=5\}\]  
main Task 100  
pool-1-thread-3 Task 6  
pool-1-thread-4 Task 7

控制台输入后继续执行：

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

pool-1-thread-3 Task 4

\[Task\{i=5\}\]

pool-1-thread-2 Task 5  
Process finished with exit code 0

### 4、singleThreadPool 单个线程池 ###

顾名思义这个线程池里面只有一个线程，这个一个线程的线程池可以保证我们扔进去的任务是顺序执行的

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /** * @program: bike-lease * @description: 单线程池 * @author: Jack.Fang * @date:2021-06-30 下午 4:05 **/
    public class TestSingleThreadPool { 
        public static void main(String[] args) { 
            ExecutorService service = Executors.newSingleThreadExecutor();
            for(int i=0;i<5;i++){ 
                final int j=i;
                service.execute(()->{ 
                    System.out.println(j+" "+Thread.currentThread().getName());
                });
            }
        }
    }

### 5、CachedPool 缓存池 ###

他没有核心线程，最大线程可以有好多好多线程，然后60秒没人理他，就回收了，他的任务列表用的是SynchronousQueue，他是用的默认工厂，拒绝策略也是默认的。

他的特点就是，你来一个任务我给你启动一个线程，当然前提是我的线程池里面有线程存在并且没有到达60秒的回收时间，来一个任务，如果有线程存在我就用现有的线程池。但是在有新的任务来的时候，如果其他线程忙我就启动一个新的。

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /** * @program: bike-lease * @description: * @author: Jack.Fang * @date:2021-06-30 下午 4:09 **/
    public class TestCachedPool { 
        public static void main(String[] args) throws InterruptedException { 
    
            ExecutorService service = Executors.newCachedThreadPool();
            System.out.println(service);
    
            for(int i=0;i<2;i++){ 
                service.execute(()->{ 
                    try{ 
                        TimeUnit.MILLISECONDS.sleep(500);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName());
                });
            }
            System.out.println(service);
            TimeUnit.SECONDS.sleep(80);
            System.out.println(service);
        }
    
    }

结果：  
java.util.concurrent.ThreadPoolExecutor@4783da3f\[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0\]  
java.util.concurrent.ThreadPoolExecutor@4783da3f\[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0\]  
pool-1-thread-2  
pool-1-thread-1

80秒后线程被回收  
java.util.concurrent.ThreadPoolExecutor@4783da3f\[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2\]

Process finished with exit code 0

### 6、FixedThreadPool ###

fixed是固定的含义，就是一个固定的一个线程数。FixedThreadPool指定一个参数，到底有多少个线程，他的最大线程数和核心线程数都是固定的就没有回收之所所以把他指定成0。

例子：  
用一个固定的线程有一个好处是什么呢，就是你可以进行并行计算，那么 **并行和并发有什么区别concurrent vs parallel**：并发是指任务提交，并行是指任务执行。并行是并发的子集。并行是多个cpu可以同时进行处理，并发是多个任务同时过来。FixedThreadPool是确实可以让你得任务来并行处理的，那么并行处理的时候就可以真真正正的提高效率。看下面这个方法isPrime判断一个数是不是质数，然后写了一个getPrime方法获取2个数之间的所有质数，这样就可以分割任务并行处理。  
我有多少cpu，这个取决于机器核数。扔进去submit他是异步的，拿到get的时候才知道你们到底有多少个，全部get完了之后相当于所有的线程都知道结果了，最后计算下时间，到底是串行快还是并行快一目了然。因此 ，这也是线程池使用的一种方式。

import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    /** * @program: bike-lease * @description: 固定线程池 * @author: Jack.Fang * @date:2021-07-02 下午 3:19 **/
    public class TestFixedThreadPool { 
    
    
        /** * 获取2个数之间的所有质数 * @param start * @param end * @return */
        static List<Integer> getPrime(int start,int end){ 
            List<Integer> result = new ArrayList<>();
            for(int i=start;i<=end;i++){ 
                if(isPrime(i)) { 
                    result.add(i);
                }
            }
            return result;
        }
    
        /** * 是否为质数 * @param num * @return */
        static boolean isPrime(int num){ 
            for(int i=2;i<=num/2;i++){ 
                if(num % i==0) { 
                    return false;
                }
            }
            return true;
        }
    
        static class MyTask implements Callable<List<Integer>>{ 
    
            int startPos,endPos;
    
            MyTask(int s, int e) { 
                this.startPos = s;
                this.endPos = e;
            }
    
            @Override
            public List<Integer> call() throws Exception { 
                return getPrime(startPos,endPos);
            }
        }
    
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
    
            //单线程算法
            long start = System.currentTimeMillis();
            getPrime(1,200000);
            long end = System.currentTimeMillis();
            System.out.println(end-start);
    
            //多线程算法
            //cpu核心数
            final int cpuCoreNum = 4;
            ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);
    
            MyTask t1 = new MyTask(1,80000);
            MyTask t2 = new MyTask(80001,130000);
            MyTask t3 = new MyTask(130001,170000);
            MyTask t4 = new MyTask(170001,200000);
    
            FutureTask<List<Integer>> f1 = (FutureTask<List<Integer>>) service.submit(t1);
            FutureTask<List<Integer>> f2 = (FutureTask<List<Integer>>) service.submit(t2);
            FutureTask<List<Integer>> f3 = (FutureTask<List<Integer>>) service.submit(t3);
            FutureTask<List<Integer>> f4 = (FutureTask<List<Integer>>) service.submit(t4);
    
            start = System.currentTimeMillis();
            f1.get();
            f2.get();
            f3.get();
            f4.get();
            end = System.currentTimeMillis();
            System.out.println(end-start);
        }
    }

运行结果：  
2366  
663

### Cache vs Fixed 怎么选择 ###

什么时候用Cache什么时候用Fixed，你得精确的控制你有多少个线程数，控制数量问题多数情况下你得预估并发量。如果线程池中的数量过多，最终他们会竞争稀缺的处理器和内存资源，浪费大量的时间在上下文切换上，反之，如果线程的数量过少，正如你得应用所面临的情况，处理器的一些核心可能就无法充分利用。线程池大小与处理器的利用率之比可以用公式来进行计算：线程池=你有多少个cpu乘以 cpu期望利用率 乘以 （1+W/C）。W除以C是等待时间与计算时间的比率。

家里你这个任务并不确定他的量平稳与否，就像是任务来的时候他可能忽高忽低，但是我要保证这个任务来时有人做这个事儿，那么我们可以用Cache,当然你要保证这个任务不会堆积。那Fixed的话就是这个任务来的比较平稳，我们大概的估算了一个值，就是这个值完全可以处理他，我就直接new这个值的线程来扔在这里就ok了。（阿里提议，自己估算，进行精确定义）

### 7、SchedulePool 任务线程池 ###

定时任务线程池，就是我们原来学过一个定时器任务，隔一段时间之后这个任务会执行。这个就是专门用来执行定时任务的一个线程池。看源码，它本质上还是ThreadPoolExecutor,还是那7个参数，专门给定时任务用的这样一个线程池罢了。

例子：  
指定核心线程数4，其实里面有一些还用的方法比如是scheduleAtFixedRate间隔多长时间在一个固定的频率上来执行一次这个任务，可以通过这样的方式灵活的对于时间上的一个控制，第一个参数Delay,第一个任务执行之前需要往后推多长时间；第二个operiod间隔多长时间；第三个参数是时间单位。

import java.util.Random;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    
    /** * @program: bike-lease * @description: 定时任务线程池 * @author: Jack.Fang * @date:2021-07-02 下午 3:57 **/
    public class TestScheduledPool { 
        public static void main(String[] args) { 
            ScheduledExecutorService service = Executors.newScheduledThreadPool(4);
            service.scheduleAtFixedRate(()->{ 
                try { 
                    TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName());
            },0,500,TimeUnit.MILLISECONDS);
        }
    }

结果：  
pool-1-thread-1  
pool-1-thread-1  
pool-1-thread-2  
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  
pool-1-thread-1  
pool-1-thread-1  
pool-1-thread-4  
pool-1-thread-4  
pool-1-thread-4  
pool-1-thread-4  
pool-1-thread-3  
…