线程进阶：生产者消费者模式和线程池

一、生产者消费者模式

这是一种不属于GOF23的设计者模式

这种模式分为三个对象：一个生产者，一个消费者，一个缓存区。

• 生产者

某些程序/进程/线程负责生产数据就属于生产者

• 消费者

某些程序/进程/线程负责使用数据就属于消费者

假设在生产者和消费者之间没有一个缓存区，那么会出现什么情况呢？

1. 耦合性高，生产者和消费者联系紧密，不利于系统的扩展和维护
2. 并发性能低，同时能处理请求量少
3. 忙闲不均，生产者和消费者的速度不一致，带来系统资源的浪费

那么如果有一个缓存区，那么两者之间的依赖性就大大降低，也就是解耦。同时生产者也可以生产多个资源放到缓存区，消费者也可以从缓存区多次得到资源，提高并发性能。阻塞队列可以很好的解决忙闲不均的问题。

阻塞队列

能够根据数据满或空的情况，自动对线程执行等待和通知

BlockingQueue 接口

• put 添加数据，达到上限会自动让线程等待

• take 取并删除数据，数据空了会自动让线程等待

实现类

ArrayBlockingQueue 类 数据结构为数组

LinkedBlockingQueue类 链表结构

我们这里举一个简单的例子解释一下这个模式：

比方说，早上我们都要去买包子吃早饭，那么包子师傅就是生产者，顾客就是消费者，蒸笼就是缓存区。大家都知道，蒸笼是可以由包子师傅自己决定有多少个，因此缓存区的最大值就可以自己进行设置。包子师傅（生产者）可以一次做多个包子（资源）放到蒸笼（缓存区）里，当有顾客（消费者）来购买包子（拿资源）就是从蒸笼里面拿取包子。蒸笼满了，包子师傅就不会做包子了，顾客也会来买包子；蒸笼空了，包子师傅就会继续做包子，顾客如果就喜欢吃他家的包子，那么就会等着，直到包子做好买完才离开。

代码如下：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BaoziShop2 {
    static class Baozi {
        //定义包子id
        private int id;

        public Baozi(int id) {
            this.id = id;
        }

        @Override
        public String toString() {
            return "包子-->" + id;
        }
    }

    public static void main(String[] args) {
        BaoziShop2 baoziShop = new BaoziShop2();
        /**
         * 阻塞队列
         * 蒸笼内包子上限是100个
         */
        BlockingQueue baoziList = new ArrayBlockingQueue(100);

        /**
         * 生产者线程
         * 当到达100个的时候，这条线程就会自动进行堵塞
         */
        new Thread(() -> {
            for (int i = 0; i < 200; i++) {
                Baozi baozi = new Baozi(baoziList.size());
                System.out.println(Thread.currentThread().getName() + "生产了" + baozi);
                try {
                    baoziList.put(baozi);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        /**
         * 消费者线程
         * 5个人买包子
        */
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                //每个人买10个包子
                for (int j = 0; j < 10; j++) {
                    try {
                        Baozi baozi = (Baozi) baoziList.take();
                        System.out.println(Thread.currentThread().getName()+"购买了"+baozi);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

二、线程池

线程池的功能与作用

线程池会保存一定量的线程，线程执行完任务后，会回到线程池中，等待下一个任务，节省系统资源，提升性能。

那为什么要使用线程池呢？

首先线程是一种宝贵的系统资源，执行完任务后会死亡，如果有大量任务需要处理，需要频繁的创建和销毁线程，造成系统性能降低。

线程池的使用

顶层接口：Executor（里面只有一个方法）

• execute(Runnable) 启动线程执行一个任务

ExecutorService

继承了Executor接口，对功能进行了增强添加

Executors

是一个工具类，能够更加方便的创建线程池

主要的方法

import java.time.LocalDateTime;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ThreadPoolDemo {

    public static void useCachedThreadPool(){
        //创建长度不限的线程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            int n = i;
            //使用线程池启动线程
            executorService.execute(()->{
                System.out.println(Thread.currentThread().getName()+"执行了任务" + n);
            });
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executorService.shutdown();
    }

    public static void useFixedThreadPool(){
        //创建长度固定的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 100; i++) {
            int n = i;
            //使用线程池启动线程
            executorService.execute(()->{
                System.out.println(Thread.currentThread().getName()+"执行了任务"+n);
            });
        }
        executorService.shutdown();
    }

    public static void useSingleThreadPool(){
        //创建单一长度的线程池
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 100; i++) {
            int n = i;
            //使用线程池启动线程
            executorService.execute(()->{
                System.out.println(Thread.currentThread().getName()+"执行了任务"+n);
            });
        }
        executorService.shutdown();
    }

    public static void useScheduledThreadPool(){
        //获得可调度的线程池
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
        //执行可调度任务
        System.out.println("------------");
        scheduledExecutorService.scheduleAtFixedRate(()->{
            System.out.println(Thread.currentThread().getName()+"---->"+ LocalDateTime.now());
        },1,3, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
//        useCachedThreadPool();
//        useFixedThreadPool();
//        useSingleThreadPool();
//        useScheduledThreadPool();
    }
}

线程池的优化配置

线程池的实现类

ThreadPoolExecutor

线程池的构造方法参数：

• corePoolSize 核心线程数，创建线程池后自带线程，不会进行销毁

• maximumPoolSize 最大线程数

• keepAliveTime 存活时间，非核心线程能够闲置的时间，超过后被销毁

• timeUnit 时间单位

• blockingQueue 阻塞队列 存放任务（Runnable）的集合