单例模式与消费者生产者模型，以及线程池的基本认识与模拟实现 ...

媒介

今天我们就来讲讲什么是单例模式与线程池的相干知识，这两个内容也是我们多线程中比力重要的内容。其次单例模式也是我们常见计划模式。

单例模式


那么什么是单例模式呢？上面说到的计划模式又是什么？
实在单例模式就是计划模式的一种。我们在学习过程中会不停编程，计划合理的代码布局和逻辑。厥后就有许多种比力常用的布局，这时就有一些大佬总结这些常用的代码布局，逻辑，并给这些布局定名成不同的模式，而这些模式就是我们所说的单例模式。那么接下来我们就来学习一下，此中之一的单例模式。

单例模式就是在某个类程序中保证某个实类对象在程序中只有一个实例，而且他的实例是不能new出来的，你只能通过其提供的getInstance方法来获取他的实例化对象。单例模式也分为两个经典的代码编写方式。“懒汉模式”和“饿汉模式”。

饿汉模式

1. class Singleton {
2.     private static Singleton instance = new Singleton();
3.     private Singleton() {}
4.     public static Singleton getInstance() {
5.         return instance;
6.     }
7. }

复制代码

如上代码：我们在要实现单例模式的类中，直接创建一个本类的成员变量，然后就是最牛的“点睛之笔”了。
我们直接将构造方法私有化，那么这时别的类就无法通过构造方法new出这个类的实例化对象了。

什么是饿汉模式呢？这里重点留意“饿”这个字，由于“饿”所以非常急，在类中直接将成员变量在定义时就直接实例化。

懒汉模式

1. class Singleton {
2.    
3.     private static volatile Singleton instance = null;
4.     private Singleton() {}
5.     public static Singleton getInstance() {
6.         if (instance == null) {
7.             synchronized (Singleton.class) {
8.                 if (instance == null) {
9.                     instance = new Singleton();
10.                 }
11.             }
12.         }
13.         return instance;
14.     }
15. }

复制代码

我们可以很清晰的看到，懒汉模式相较于饿汉模式代码多了些，由于懒汉模式在多线程下需要加锁，如果不加锁可能有多个线程同时调用getInstance（）方法就有可能创建出多个实例，这就不符合单例模式的设定了。而且还加上了volatile 保证了内存的可见性也是防止线程安全的发生。

如何理解两层 if判断 ？
最里层的 if ，判断该类是否被实例化，如果没有实例化即 ==null 我们就new一个对象并返回，如果已经实例化过就直接返回。

最外层的 if 我们知道加锁和开锁也是一个开销比力高的事变，我们就要经只管减少加锁开锁的次数，当我们的实例已经创建了，我们在就可以直接返回了，但是如果不加这一层 if，程序就会加锁判断部分，这就导致无用开销，所以我们就可以再加一层判断。这一层就是为了防止无用的加锁。


壅闭队列


什么是壅闭队列呢？
正如字面他是一个可以壅闭的队列，他跟普通队列一样有着FIFO（“先辈先出”）出的性质。但是他会在两种情况下发生壅闭。

• 当队列满时，如果另有元素要入队列那么就会发生壅闭。
  
• 当队列为空时，如果有出队列的哀求那么也会发生壅闭。
  

壅闭队列的一个经典应用场景就是“生产者消费者模型”，这也是一个非常典型的开发模型。


生产者消费者模型

生产者消费者模型就是运用一个容器来解决生产者和消费者的强耦合的题目。
生产者和消费者之间不直接通讯，通过壅闭队列来进行通讯，所以生产者生产完数据后不在需要等待消费者处置惩罚，而是直接将数据丢给壅闭队列，消费者也不找生产者要数据，而是直接从壅闭队列中取数据。如许就成功的对生产者和消费者进行解耦。

除了解耦，壅闭队列还起到了缓冲区的作用，壅闭队列就均衡了生产者和消费者的处置惩罚本领，起到了消峰填谷。好比在某些购物日，或秒杀抢购的情况下，如果没有壅闭队列，突然暴增的哀求，如果让服务器直接去处置惩罚，我们的服务器有可能会处置惩罚不外来，而导致奔溃，但是如果有了壅闭队列我们就可以把哀求放进壅闭队列中，再由消费者线程逐步处置惩罚订单。


Java标准库中的壅闭队列


在Java标准库中我们有壅闭队列 BlockingQueue（这是一个接口），此中他的实现类是LinkedBlockingQueue
此中这个队列中我们有put（）方法表示入队列，另有take（）出队列，者两个方法是具有壅闭功能的.
当然这个队列也有offer，poll，peek，等方法，但这都是不具有壅闭功能的。

下面我们通过编写一段代码，象形的展示了壅闭队列的壅闭功能，非常直观。
我们先是创建第一个消费者线程，让其不停的从壅闭队列中去数据，但刚开始壅闭队列中没有数据，所以他就会进行壅闭，所以我们创建了第二个线程（生产者线程）我们每隔一段时间生成一个数据，并放进队列中，这时生产者线程就可会马大将队列里的数据给取出来了。

1. import java.util.Random;
2. import java.util.concurrent.BlockingQueue;
4. import java.util.concurrent.LinkedBlockingQueue;
6. public class Main {
7.     public static void main(String[] args) throws InterruptedException {
8.         BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>();
9.         Thread customer = new Thread(() -> {
10.             while (true) {
11.                 try {
12.                     int value = blockingQueue.take();
13.                     System.out.println("消费元素: " + value);
14.                 } catch (InterruptedException e) {
15.                     e.printStackTrace();
16.                 }
17.             }
18.         }, "消费者");
19.         customer.start();
20.         Thread producer = new Thread(() -> {
21.             Random random = new Random();
22.             while (true) {
23.                 try {
24.                     int num = random.nextInt(1000);
25.                     System.out.println("⽣产元素: " + num);
26.                     blockingQueue.put(num);
27.                     Thread.sleep(1000);
28.                 } catch (InterruptedException e) {
29.                     e.printStackTrace();
30.                 }
31.             }
32.         }, "⽣产者");
33.         producer.start();
34.         customer.join();
35.         producer.join();
36.     }
37. }

复制代码

壅闭队列的模拟实现


这里主要通过wait（）进行壅闭，当我们发现队列满时，或空时，我们的put方法和take（）方法就要进行壅闭，也就是调用wait（）方法以及我们的synchronize，但是需要留意的是我们这里的判断只管还是用while循环来进行判断，由于在我们notifyAll 的时候, 该线程从 wait 中被唤醒,但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能⼜已经队列满了

• 假设队列初始是满的，生产者 P1 调用 wait() 并开释锁。
  
• 消费者 C1 抢到锁，消费一个数据，调用 notifyAll()，唤醒 P1。
  
• 但此时可能有另一个生产者 P2 争先抢到锁，并插入数据，导致队列又满了。
  
• 如果 P1 用 if，它会直接执行 items[tail] = value（由于之前检查 size == items.length 是 true，但被唤醒后未重新检查），导致队列溢出。
  
• 用 while 会重新检查条件，发现队列还是满的，继续 wait()。
  

如下代码：我们通过synchroniz关键字和wait（）方法，完成了壅闭队列。

1. class BlockingQueue1{
2.     private int[] items = new int[1000];
3.     private volatile int size = 0;
4.     private volatile int head = 0;
5.     private volatile int tail = 0;
6.     public void put(int value) throws InterruptedException {
7.         synchronized (this) {
8.             // 此处最好使⽤ while.
9.             // 否则 notifyAll 的时候, 该线程从 wait 中被唤醒,
10.             // 但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能⼜已经队列满了
11.             // 就只能继续等待
12.             while (size == items.length) {
13.                 wait();
14.             }
15.             items[tail] = value;
16.             tail = (tail + 1) % items.length;
17.             size++;
18.             notifyAll();
19.         }
20.     }
21.     public int take() throws InterruptedException {
22.         int ret = 0;
23.         synchronized (this) {
24.             while (size == 0) {
25.                 wait();
26.             }
27.             ret = items[head];
28.             head = (head + 1) % items.length;
29.             size--;
30.             notifyAll();
31.         }
32.         return ret;
33.     }
34.     public synchronized int size() {
35.         return size;
36.     }
37.     // 测试代码
38.     public static void main(String[] args) throws InterruptedException {
39.         BlockingQueue1 blockingQueue = new BlockingQueue1();
40.         Thread customer = new Thread(() -> {
41.             while (true) {
42.                 try {
43.                     int value = blockingQueue.take();
44.                     System.out.println(value);
45.                 } catch (InterruptedException e) {
46.                     e.printStackTrace();
47.                 }
48.             }
49.         }, "消费者");
50.         customer.start();
51.         Thread producer = new Thread(() -> {
52.             Random random = new Random();
53.             while (true) {
54.                 try {
55.                     blockingQueue.put(random.nextInt(10000));
56.                 } catch (InterruptedException e) {
57.                     e.printStackTrace();
58.                 }
59.             }
60.         }, "⽣产者");
61.         producer.start();
62.         customer.join();
63.         producer.join();
64.     }
65. }

复制代码

定时器

定时器是什么呢？就是一个可以根据时间定时执行任务的容器吧
定时器的主要构成：

• 一个带优先级的队列（ (不要使⽤ PriorityBlockingQueue, 容易死锁 ）
  
• 此中队列中的每一个元素是一个Task
  
• Task中存在一个带有时间属性，此中队首元素是即将执行的元素。
  
• 另有存在一个工作线程worker不停扫描队首元素,,检查时间是否以及到了，是否开始执行
  

定时器的模拟实现：
首先我们先对MyTask重写我们的compareTo方法，如果不在这里重写，就要在创建队列的时候对其构造方法传入一个比力器的参数即（Comparable<MyTask>）。

且MyTask中是定义一个long类型的属性time，我们就可以使用时间戳来表示他要在何时执行任务。
然后定义构造方法，此中包罗两个参数，第一个就是当前的时间戳，第二个是多少毫秒后执行当前任务。

然后我们在提交任务去定时器时，只需要传入这两个参数给schedule即可。在方法内部我们会先根据这两个参数
构造出一个Mytask,然后放进优先级队列中。

1. class MyTask implements Comparable<MyTask> {
2.     public Runnable runnable;
3.     // 为了⽅便后续判定, 使⽤绝对的时间戳.
4.     public long time;
5.     public MyTask(Runnable runnable, long delay) {
6.         this.runnable = runnable;
7.         // 取当前时刻的时间戳 + delay, 作为该任务实际执⾏的时间戳
8.         this.time = System.currentTimeMillis() + delay;
9.     }
10.     @Override
11.     public int compareTo(MyTask o) {
12.         // 这样的写法意味着每次取出的是时间最⼩的元素.
13.         // 到底是谁减谁?? 俺也记不住!!! 随便写⼀个, 执⾏下, 看看效果~~
14.         return (int)(this.time - o.time);
15.     }
16. }
18. class MyTimer {
19.     // 核⼼结构
20.     private PriorityQueue<MyTask> queue = new PriorityQueue<>();
21.     // 创建⼀个锁对象
22.     private Object locker = new Object();
23.     public void schedule(Runnable command, long after) {
24.         // 根据参数, 构造 MyTask, 插⼊队列即可.
25.         synchronized (locker) {
26.             MyTask myTask = new MyTask(command, after);
27.             queue.offer(myTask);
28.             locker.notify();
29.         }
30.     }
32.     // 在这⾥构造线程, 负责执⾏具体任务了.
34.     public MyTimer() {
35.         Thread t = new Thread(() -> {
36.             while (true) {
37.                 try {
38.                     synchronized (locker) {
39.                         // 阻塞队列, 只有阻塞的⼊队列和阻塞的出队列, 没有阻塞的查看队⾸元素.
40.                         while (queue.isEmpty()) {
41.                             locker.wait();
42.                         }
43.                         MyTask myTask = queue.peek();
44.                         long curTime = System.currentTimeMillis();
45.                         if (curTime >= myTask.time) {
46.                             // 时间到了, 可以执⾏任务了
47.                             queue.poll();
48.                             myTask.runnable.run();
49.                         } else {
50.                             // 时间还没到
51.                             locker.wait(myTask.time - curTime);
52.                         }
53.                     }
54.                 } catch (InterruptedException e) {
55.                     e.printStackTrace();
56.                 }
57.             }
58.         });
59.         t.start();
60.     }
62.    
63. }

复制代码

线程池

线程池是什么？通俗来讲就是先创建好一些线程，当我们需要创建线程时，直接从线程池里取即可，不需要在创建，而且用完后直接将线程返回给线程池，如许就减少了我们创建和烧毁线程的开销。


Java标准库中的线程池


Executors是一个工厂类，提供了创建各种类型线程池的静态方法

固定巨细线程池

• ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads);
  
• 固定命目的线程
  
• 无界任务队列(LinkedBlockingQueue)
  
• 适用于负载较重的服务器
  

单线程线程池

• ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
  
• 只有一个工作线程
  
• 保证任务次序执行
  
• 无界任务队列
  

可缓存线程池

• ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
  
• 线程数目根据需要自动调解
  
• 空闲线程60秒后回收
  
• 适用于执行大量短期异步任务
  

定时任务线程池

• ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(int corePoolSize);
  
• 核心线程数固定，非核心线程数无限定
  
• 支持定时及周期性任务执行
  

如下代码就可以创建一个含有10个线程的线程池，此中
ExecutorService executorService = Executors.newFixedThreadPool(10);

从上上面代码可以很容易看出Executors.newFixedThreadPool(10)的返回值时一个ExecutorService，然后我们可以往线程池里提交任务执行了。
如下截图：我们调用submit方法时，只需要传入一个Runable的对象即可，跟创建线程的方法相似。

1. mport java.util.concurrent.ExecutorService;
2. import java.util.concurrent.Executors;
4. public class Main {
5.     public static void main(String[] args) {
6.         ExecutorService executorService = Executors.newFixedThreadPool(10);
7.         executorService.submit(()->System.out.println("一个人任务"));
8.     }
9. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。