Java常用六种线程池

深碍√TFBOYSˉ_ 2023-06-14 10:54 17阅读 0赞

## ***欢迎大家访问我的个人博客***：[L\_SKH’Blog][L_SKH_Blog] ##

# 一、FixedThreadPool #

固定数量线程池

package org.skh.c026;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 15:08
     */
    public class T05_ThreadPool {
        
        public static void main(String[] args) throws InterruptedException {
        
            //固定数量的线程池
            ExecutorService service = Executors.newFixedThreadPool(5);
            //向五个线程提交6个任务
            //因为是固定数量 所以不会有第六个线程 也所以某个线程会执行两个任务
            for (int i=0;i<6;i++){
        
                service.execute(()->{
        
                    try {
        
                        TimeUnit.MILLISECONDS.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName());
    
                });
            }
    
            System.out.println(service); //打印线程池信息
            /**
             * 状态: Running,
             * 线程池大小: pool size = 5,
             * 存活的线程数: active threads = 5,
             * 等待的任务的数量: queued tasks = 1,
             * 完成的任务数量: completed tasks = 0
             */
            service.shutdown();  //关闭线程池 但不会立即关闭只会返回一个true 要等待所有任务完成
            System.out.println(service.isTerminated()); //是否停止
            System.out.println(service.isShutdown()); //是否关闭 返回true但没有真的关闭 只是一个flag
            System.out.println(service);
            TimeUnit.SECONDS.sleep(5);
            System.out.println(service.isTerminated());
            System.out.println(service.isShutdown()); //此时才是真的关闭了 因为任务已全部完成
            System.out.println(service);
        }
    }

# 二、CachedPool #

与之前的线程池不同 这个是只有提交了任务时才会启动一个线程 一开始是没有线程的 来一个任务就开启一个线程 当然前提是线程池里没有空闲的并且存活的线程 另外如果一个线程如果在60s内没有被使用 则会被kill

package org.skh.c026;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 20:59
     */
    public class T08_CachedPool {
        
        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);
        }
    }

# 三、SingleThreadPool #

顾名思义，只有一个线程处理任务的线程，线程池中只有一个线程 所以下面程序中的五个任务只有一个线程执行。所以可以保证任务执行的顺序性

package org.skh.c026;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 21:22
     */
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    public class T09_SingleThreadPool {
        
    
        public static void main(String[] args) {
        
            ExecutorService service = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 5; i++) {
        
               int j = i;
                service.execute(() -> {
        
                    System.out.println(j + " " + Thread.currentThread().getName());
                });
            }
        }
    }

# 四、ScheduledPool #

ScheduledPool Scheduled: 计划中的,定时的。执行定时的任务,类似Delay, 可以替代Timer。

package org.skh.c026;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 21:27
     */
    
    import java.util.Random;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    public class T10_ScheduledPool {
        
    
        public static void main(String[] args) {
        
            ScheduledExecutorService service = Executors.newScheduledThreadPool(4);
            // 使用固定的频率执行某个任务
            // 四个参数
            // command: 执行的任务
            // initialDelay: 第一次执行延时多久执行
            // period: 每隔多久执行一次这个任务
            // unit: 时间单位
            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);  // 每隔500ms打印一下线程名称
            //每经过500ms 会在执行一次任务 最大线程数量为4个 如果sleep的时间大于500会启用另一个线程来执行
            //当所有的四个线程都在执行任务时 会等待某个线程空闲下来来执行它
        }
    }

# 五、WorkStealingPool #

顾名思义，每个线程都有自己维护的队列，当一个线程处理完自己的队列后，会去‘偷’别人的任务队列进行处理。

package org.skh.c026;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 21:42
     */
    
    import java.io.IOException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    public class T11_WorkStealingPool {
        
    
        public static void main(String[] args) throws IOException {
        
            // CPU 核数
            System.out.println(Runtime.getRuntime().availableProcessors());
    
            // workStealingPool 会自动启动cpu核数个线程去执行任务
            ExecutorService service = Executors.newWorkStealingPool();
            service.execute(new R(1000));  // 我的cpu核数为12 启动13个线程,其中第一个是1s执行完毕,其余都是2s执行完毕,
            // 有一个任务会进行等待,当第一个执行完毕后,会再次偷取第十三个任务执行
            for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
        
                service.execute(new R(2000));
            }
    
            // 因为work stealing 是deamon线程,即后台线程,精灵线程,守护线程
            // 所以当main方法结束时, 此方法虽然还在后台运行,但是无输出
            // 可以通过对主线程阻塞解决
            System.in.read();
        }
    
        static class R implements Runnable {
        
    
            int time;
    
            R(int time) {
        
                this.time = time;
            }
    
            @Override
            public void run() {
        
                try {
        
                    TimeUnit.MILLISECONDS.sleep(time);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "  " + time);
            }
        }
    }

# 六、ForkJoinPool #

将一个任务拆分多个任务执行(可以无限切分),然后将结果合并

package org.skh.c026;
    
    /**
     * @Created IntelliJ IDEA.
     * @Author L_SKH
     * @Date 2019/11/18 21:54
     */
    
    import java.io.IOException;
    import java.util.Arrays;
    import java.util.Random;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveAction;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * T12_ForkJoinPool  分而治之
     * Fork: 分叉
     * Join: 合并
     * <p>
     * 将一个任务拆分多个任务执行(可以无限切分),然后将结果合并
     * <p>
     * 比如大量的并行计算, 如下: 求100_0000个数字之和, 使用多线程
     */
    public class T12_ForkJoinPool {
        
    
        static int[] nums = new int[100_0000];
        static final int MAX_NUM = 5_0000; // 每个线程最多可以运行5万个数字相加
        static Random random = new Random();
    
        // 初始化这100_000个数字, 每个数字范围在100之内
        static {
        
    
            for (int i = 0; i < nums.length; i++) {
        
                nums[i] = random.nextInt(100);
            }
            // 所有数字和, 事先计算:
            System.out.println(Arrays.stream(nums).sum()); // 使用单线程stream api 进行求和
        }
    
        /**
         * RecursiveAction(AddTask): 递归操作 没有返回值
         * RecursiveTask(AddTask2): 递归操作,有返回值
         */
    
        static class AddTask extends RecursiveAction {
        
    
            int start, end;
    
            AddTask(int start, int end) {
        
                this.start = start;
                this.end = end;
            }
    
            @Override
            protected void compute() {
        
    
                // 进行计算
                // 如果计算的数的和的范围 小于 MAX_NUM, 进行计算,否则进行 fork
                if (end - start <= MAX_NUM) {
        
                    long sum = 0;
                    for (int i = start; i < end; i++) {
        
                        sum += nums[i];
                    }
                    System.out.println("sum = " + sum);
                } else {
        
                    int middle = (end - start) / 2;
                    AddTask subTask1 = new AddTask(start, middle);
                    AddTask subTask2 = new AddTask(middle, end);
                    subTask1.fork();
                    subTask2.fork();
                }
            }
        }
    
        static class AddTask2 extends RecursiveTask<Long> {
        
    
            int start, end;
    
            AddTask2(int start, int end) {
        
                this.start = start;
                this.end = end;
            }
    
            @Override
            protected Long compute() {
        
                // 进行计算
                // 如果计算的数的和的范围 小于 MAX_NUM, 进行计算,否则进行 fork
                if (end - start <= MAX_NUM) {
        
                    long sum = 0;
                    for (int i = start; i < end; i++) {
        
                        sum += nums[i];
                    }
                    return sum;
                } else {
        
                    int middle = start + (end - start) / 2; // 注意这里，如果有问题，会抛出java.lang.NoClassDefFoundError: Could not initialize class java.util.concurrent.locks.AbstractQueuedSynchronizer$Node 异常
                    AddTask2 subTask1 = new AddTask2(start, middle);
                    AddTask2 subTask2 = new AddTask2(middle, end);
                    subTask1.fork();
                    subTask2.fork();
                    //有返回值
                    return subTask1.join() + subTask2.join();
                }
            }
        }
    
        // 运行
        public static void main(String[] args) throws IOException {
        
            ForkJoinPool fjp = new ForkJoinPool();
    
            AddTask2 task = new AddTask2(0, nums.length);
    
            fjp.execute(task);
    
            System.out.println(task.join());
    
            System.in.read();
        }
    
    }

[L_SKH_Blog]: http://47.103.221.24/