多线程-- 线程池使用之等待所有任务执行完和关闭线程池

【一】自定义线程池

（1）为什么使用线程池
每一个线程的启动和结束都是比较消耗时间和占用资源的。使用线程池的过程中创建固定数量的线程，不用创建多余新的线程，而是循环使用那些已经存在的线程。

（2）自定义线程池设计思路
1-准备一个任务容器
2-一次性启动10个消费者线程
3-刚开始任务容器是空的，所以线程都wait在上面
4-直到一个外部线程往这个任务容器中扔了一个“任务”，就会有一个消费者线程被唤醒
5-这个消费者线程取出“任务”，并且执行这个任务，执行完毕后，继续等待下一次任务的到来
6-如果短时间内，有较多的任务加入，name就会有多个线程被唤醒，去执行这些任务

public class ThreadPool {
  
    // 线程池大小
    int threadPoolSize;
  
    // 任务容器
    LinkedList<Runnable> tasks = new LinkedList<Runnable>();
  
    // 试图消费任务的线程
  
    public ThreadPool() {
        threadPoolSize = 10;
  
        // 启动10个任务消费者线程
        synchronized (tasks) {
            for (int i = 0; i < threadPoolSize; i++) {
                new TaskConsumeThread("任务消费者线程 " + i).start();
            }
        }
    }
  
    public void add(Runnable r) {
        synchronized (tasks) {
            tasks.add(r);
            // 唤醒等待的任务消费者线程
            tasks.notifyAll();
        }
    }
  
    class TaskConsumeThread extends Thread {
        public TaskConsumeThread(String name) {
            super(name);
        }
  
        Runnable task;
  
        public void run() {
            System.out.println("启动： " + this.getName());
            while (true) {
                synchronized (tasks) {
                    while (tasks.isEmpty()) {
                        try {
                            tasks.wait();
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                    task = tasks.removeLast();
                    // 允许添加任务的线程可以继续添加任务
                    tasks.notifyAll();
  
                }
                System.out.println(this.getName() + " 获取到任务，并执行");
                task.run();
            }
        }
    }
  
}

public class TestThread {
       
    public static void main(String[] args) {
        ThreadPool pool = new ThreadPool();
  
        for (int i = 0; i < 5; i++) {
            Runnable task = new Runnable() {
                @Override
                public void run() {
                    //System.out.println("执行任务");
                    //任务可能是打印一句话
                    //可能是访问文件
                    //可能是做排序
                }
            };
             
            pool.add(task);
             
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
  
    }
           
}

【二】java自带的线程池

（1）线程池使用实例–使用java线程池实现穷举法破解密码

主要方法类

public class PassWordThreadpool {

    //第一步：把穷举法生成密码并且进行匹配的方法写好
    //如果匹配到密码了就停止遍历
    private boolean found = false;

    public synchronized void generatePwd(char[] guessPwd,String pwd, List<String> pwdList) {
        generatePwd(guessPwd,0,pwd, pwdList);
    }

    public synchronized void generatePwd(char[] guessPwd,int index,String pwd, List<String> pwdList) {
        //遍历数值和字母来生成密码
        if(found){
            return;
        }
        for (short i='0';i<'z';i++) {
            if(!Character.isLetterOrDigit(i)){
                continue;
            }
            char c = (char) i;
            guessPwd[index] = c;
            if(index==pwd.length()-1){
                //把三个字母的数组拼接成字符串
                String guessResult = new String(guessPwd);
                pwdList.add(guessResult);
                if(guessResult.equals(pwd)){
                    System.out.println("密码找到了，是："+guessResult);
                    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    System.out.println("当前时间是："+df.format(new Date()));
                    found = true;
                    return;
                }
            } else {
                generatePwd(guessPwd,index+1,pwd, pwdList);
            }
        }
    }
}

线程池使用

public class TestThreadpoolGuesspwd {
    public static void main(String[] args) throws InterruptedException {

        String pwd = randomPwd(3);
        System.out.println("生成的密码是："+pwd);
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("当前时间是："+df.format(new Date()));
        List<String> pwdList = new CopyOnWriteArrayList<>();
        PassWordThreadpool passWordThread = new PassWordThreadpool();

        // 创建线程池
        ThreadPoolExecutor threadPool= new ThreadPoolExecutor(10, 15, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

        // 把任务提交到线程池
        threadPool.execute(() -> {
            char[] guessPwd = new char[pwd.length()];
            passWordThread.generatePwd(guessPwd,pwd, pwdList);
        });

    }

    public static String randomPwd(int length) {
        String pool = "";
        for (short i = '0'; i <= '9'; i++) {
            pool += (char) i;
        }
        for (short i = 'a'; i <= 'z'; i++) {
            pool += (char) i;
        }
        for (short i = 'A'; i <= 'Z'; i++) {
            pool += (char) i;
        }
        char cs[] = new char[length];
        for (int i = 0; i < cs.length; i++) {
            int index = (int) (Math.random() * pool.length());
            cs[i] = pool.charAt(index);
        }
        String result = new String(cs);
        return result;
    }
}

【三】如何优雅的等待线程池所有任务执行完

（1）案例一：主线程在子线程结束前先结束了

public class TestThreadpoolImport01 {

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(0,25,100L,
            TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) {
        Student student = null;
        List<Student> studentList = new ArrayList<>();
        int age = 0;
        int heigh = 0;
        //添加50万个数据
        for (int i=0;i<=500000;i++) {
            age = (int) Math.floor((Math.random() * 10) + 20);
            heigh = (int) Math.floor((Math.random() * 160) + 180);
            student = new Student("name"+i,age,heigh,"玩");
            studentList.add(student);
        }

        long start = System.currentTimeMillis();
        for (Student student1:studentList) {
            //模拟对数据信息进行二次处理
            executor.submit(()->student1.setName(student1.getName()+"这是后缀"));
        }
        long end = System.currentTimeMillis();
        System.out.println("添加数量："+studentList.stream().filter(x->x.getName().contains("这是后缀")).count());
        System.out.println("花费时间："+(end-start));


        //输出的数据是
        // 添加数量：371014
        // 花费时间：114
        //不是50w，这是由于线程池中的子线程任务没有执行完，而主线程已经开始执行业务代码，导致成功数量变少。

    }
}

以前在没有使用线程池之前，可以将所有线程放进线程数组，然后遍历数组，给每个线程对象调用join方法，现在来探索线程池的如何实现

（2）案例二：使用CountDownLatch

public class TestThreadpoolImport02 {

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(0,25,100L,
            TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) {
        Student student = null;
        List<Student> studentList = new ArrayList<>();
        int age = 0;
        int heigh = 0;
        //添加50万个数据
        for (int i=0;i<500000;i++) {
            age = (int) Math.floor((Math.random() * 10) + 20);
            heigh = (int) Math.floor((Math.random() * 160) + 180);
            student = new Student("name"+i,age,heigh,"玩");
            studentList.add(student);
        }

        long start = System.currentTimeMillis();

        //在线程池执行之前，给计数器指定数值（与要执行代码的次数一致）也就是students.size()
        CountDownLatch countDownLatch = new CountDownLatch(studentList.size());

        for (Student student1:studentList) {
            //模拟对数据信息进行二次处理
            executor.submit(()->{
                try {
                    student1.setName(student1.getName()+"这是后缀");
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //每执行一次数值减少一
                    countDownLatch.countDown();
                    //也可以给await()设置超时时间，如果超过300s（也可以是时，分）则不再等待，直接执行下面代码。
                    //countDownLatch.await(300,TimeUnit.SECONDS);
                }
            });
        }
        try {
            //等待计数器归零
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("添加数量："+studentList.stream().filter(x->x.getName().contains("这是后缀")).count());
        System.out.println("花费时间："+(end-start));


        //输出的数据是
        // 添加数量：500000
        // 花费时间：334
        //是50w，主线程等待所有子线程执行结束后才结束

    }
}

这是一个计数器操作，在线程池执行之前，给计数器指定数值（与要执行代码的次数一致）也就是students.size()，在线程池执行代码体里面要加上countDownLatch.countDown();代表每执行一次数值减少一，最后在循环体外边写上countDownLatch.await();代表等待计数器归零。

也可以给await()设置超时时间。如果超过300s（也可以是时，分）则不再等待，直接执行下面代码。

countDownLatch.await(300,TimeUnit.SECONDS);

（3）使用Future.get()

public class TestThreadpoolImport03 {

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(0,25,100L,
            TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) {
        Student student = null;
        List<Student> studentList = new ArrayList<>();
        int age = 0;
        int heigh = 0;
        //添加50万个数据
        for (int i=0;i<500000;i++) {
            age = (int) Math.floor((Math.random() * 10) + 20);
            heigh = (int) Math.floor((Math.random() * 160) + 180);
            student = new Student("name"+i,age,heigh,"玩");
            studentList.add(student);
        }

        long start = System.currentTimeMillis();

        List<Future> futureList = new ArrayList<>();

        for (Student student1:studentList) {
            //使用submit提交会有一个返回值
            Future future = executor.submit(()->{
                try {
                    //模拟对数据信息进行二次处理
                    student1.setName(student1.getName()+"这是后缀");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            futureList.add(future);
        }

        for (Future future:futureList) {
            try {
                //监听线程池子线程执行状态及执行结果。
                future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        long end = System.currentTimeMillis();
        System.out.println("添加数量："+studentList.stream().filter(x->x.getName().contains("这是后缀")).count());
        System.out.println("花费时间："+(end-start));


        //输出的数据是
        // 添加数量：500000
        // 花费时间：334
        //是50w，主线程等待所有子线程执行结束后才结束

    }
}

使用submit提交线程，会返回一个Future的值，将每个返回值放进List，然后遍历调用get方法：future.get();

（4）使用shutdown方法
如果线程池是方法内部创建的，可以直接使用shutdown（）也会等待线程池的执行结果。同时会关闭线程池资源。

public class TestThreadpoolImport04 {

    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(0,25,100L,
                TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadPoolExecutor.CallerRunsPolicy());

        Student student = null;
        List<Student> studentList = new ArrayList<>();
        int age = 0;
        int heigh = 0;
        //添加50万个数据
        for (int i=0;i<500000;i++) {
            age = (int) Math.floor((Math.random() * 10) + 20);
            heigh = (int) Math.floor((Math.random() * 160) + 180);
            student = new Student("name"+i,age,heigh,"玩");
            studentList.add(student);
        }

        long start = System.currentTimeMillis();
        for (Student student1:studentList) {
            //模拟对数据信息进行二次处理
            executor.submit(()->student1.setName(student1.getName()+"这是后缀"));
        }
        executor.shutdown();
        try {
            executor.awaitTermination(300,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("添加数量："+studentList.stream().filter(x->x.getName().contains("这是后缀")).count());
        System.out.println("花费时间："+(end-start));


        //输出的数据是
        // 添加数量：500000
        // 花费时间：281
        //不是50w，这是由于线程池中的子线程任务没有执行完，而主线程已经开始执行业务代码，导致成功数量变少。

    }
}

【四】如何优雅的关闭线程池

【五】案例一：用线程池异步查询订单和收货地址

（1）使用两个不同的线程分别查询订单和收货地址

/**
 * @ClassName: TestThreadSearch
 * @Author: AllenSun
 * @Date: 2022/3/21 下午10:57
 */
public class TestThreadSearch {

    public static void getOrder() {
        try {
            Thread.sleep(1000);
            System.out.println("获取订单信息");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void getAddress() {
        try {
            Thread.sleep(1000);
            System.out.println("获取地址信息");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        getOrder();
        getAddress();
        long end = System.currentTimeMillis();
        //花费时间：2007
        System.out.println("花费时间："+(end-start));

    }
}

耗时结果为

获取订单信息
获取地址信息
花费时间：2008

Process finished with exit code 0

（2）使用线程池改造

/**
 * @ClassName: TestThreadSearch
 * @Author: AllenSun
 * @Date: 2022/3/21 下午10:57
 */
public class TestThreadPoolSearch {

    public static void getOrder() {
        try {
            Thread.sleep(1000);
            System.out.println("获取订单信息");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void getAddress() {
        try {
            Thread.sleep(1000);
            System.out.println("获取地址信息");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ThreadPoolExecutor executor= new ThreadPoolExecutor(
                5,
                10,
                15,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        long start = System.currentTimeMillis();
        //异步执行
        CompletableFuture<Void> addressFuture = CompletableFuture.runAsync(()->{
            getAddress();
        },executor);
        //异步执行
        CompletableFuture<Void> orderFuture = CompletableFuture.runAsync(()->{
            getOrder();
        },executor);

        //等待完成
        CompletableFuture.allOf(addressFuture,orderFuture).get();


        long end = System.currentTimeMillis();

        //花费时间：1053
        System.out.println("花费时间："+(end-start));

        executor.shutdown();

    }
}

耗时结果为

获取地址信息
获取订单信息
花费时间：1127

Process finished with exit code 0

（3）使用线程池改造

/**
 * @ClassName: TestThreadSearch
 * @Author: AllenSun
 * @Date: 2022/3/21 下午10:57
 */
public class TestThreadPoolSearch2 {
    private static final CountDownLatch ctl = new CountDownLatch(2);
    
    public static void getOrder() {
        try {
            Thread.sleep(1000);
            System.out.println("获取订单信息");
            ctl.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void getAddress() {
        try {
            Thread.sleep(1000);
            System.out.println("获取地址信息");
            ctl.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ThreadPoolExecutor executor= new ThreadPoolExecutor(
                5,
                10,
                15,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        long start = System.currentTimeMillis();
        //异步执行
        executor.execute(() -> {
            getOrder();
        });
        executor.execute(() -> {
            getAddress();
        });

        //等待完成
        try {
            //等待计数器归零
            ctl.await(20, TimeUnit.SECONDS);//最多等待20秒，不管子线程完没完
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        

        long end = System.currentTimeMillis();

        //花费时间：1053
        System.out.println("花费时间："+(end-start));

        executor.shutdown();

    }
}

耗时结果为

获取地址信息
获取订单信息
花费时间：1100

Process finished with exit code 0

【六】案例二：线程池模拟批量导入数据

public class TestThreadpoolImport02 {

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(0,25,100L,
            TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) {
        Student student = null;
        List<Student> studentList = new ArrayList<>();
        int age = 0;
        int heigh = 0;
        //添加50万个数据
        for (int i=0;i<500000;i++) {
            age = (int) Math.floor((Math.random() * 10) + 20);
            heigh = (int) Math.floor((Math.random() * 160) + 180);
            student = new Student("name"+i,age,heigh,"玩");
            studentList.add(student);
        }

        long start = System.currentTimeMillis();

        //在线程池执行之前，给计数器指定数值（与要执行代码的次数一致）也就是students.size()
        CountDownLatch countDownLatch = new CountDownLatch(studentList.size());

        for (Student student1:studentList) {
            //模拟对数据信息进行二次处理
            executor.submit(()->{
                try {
                    student1.setName(student1.getName()+"这是后缀");
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //每执行一次数值减少一
                    countDownLatch.countDown();
                    //也可以给await()设置超时时间，如果超过300s（也可以是时，分）则不再等待，直接执行下面代码。
                    //countDownLatch.await(300,TimeUnit.SECONDS);
                }
            });
        }
        try {
            //等待计数器归零
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("添加数量："+studentList.stream().filter(x->x.getName().contains("这是后缀")).count());
        System.out.println("花费时间："+(end-start));


        //输出的数据是
        // 添加数量：500000
        // 花费时间：334
        //是50w，主线程等待所有子线程执行结束后才结束

    }
}
# java
Logo
向您推荐>>Eolink开发者社区

权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区

更多推荐

深入理解 Mocha 测试框架：从零实现一个 Mocha

前言什么是自动化测试自动化测试在很多团队中都是Devops环节中很难执行起来的一个环节，主要原因在于测试代码的编写工作很难抽象，99%的场景都需要和业务强绑定，而且写测试代码的编写工作量往往比编写实际业务代码的工作量更多。在一些很多业务场景中投入产出比很低，适合写自动化测试的应该是那些中长期业务以及一些诸如组件一样的基础库。自动化测试是个比较大的概念，其中分类也比较多，比如单元测试，端对端测试，集

avatar 云原生

ELK实现containerd的容器日志采集展示【基于logging的全栈监测】

企业级ELK Stack构建介绍

avatar 云原生

(20200916 Solved)docker-compose up创建容器自动退出

问题描述如题，创建容器后自动退出了。并且docker start container无效解决方案原因是缺失了控制终端的配置，需要在docker-compose.yml中增加tty:true ，有时候这样也不行，需要再增加一个command:/bin/bash，命令不一定是这个，需要是一个不会退出的命令，然后用-d后台启动容器。Referencesdocker-compose启动容器后自动退出...

avatar 云原生