【线程】Java多线程代码案例（1）

一、“单例模式” 的实现

“单例模式”即在一个Java 进程中，要求指定的类，只能有一个实例。通过一些特殊的技巧，来确保这里的实例不会有多个。

1.1“饿汉模式”

class Singleton{//在这个类被加载的时候，初始化这个静态成员private static Singleton instance=new Singleton();public static Singleton getInstance() {return instance;}//构造方法私有，在类外部不允许new新的实例private Singleton(){}
}

1. 在类外不允许创建新的实例：
   
2. instance实例时Singleton类的静态成员变量，在内存中只有一份存储空间，是唯一的。

1.2 “懒汉模式”

“懒汉模式”创建实例的时机会相对晚一些，只有到第一次使用的时候，才会创建实例。

class Singleton{private static Singleton instance=null;public static Singleton getInstance() {if (instance==null) {instance=new Singleton();}return instance;}//构造方法私有，在类外部不允许new新的实例private Singleton(){}}

同样，在类外不允许创建新的实例。但是在类内创建实例的时机会相对晚一些。

1.3 线程安全问题

再次观察这两份代码，是否会存在线程安全问题呢？

其实在“饿汉模式”下，并不存在线程安全的问题，因为此处在加载类的时候，就创建了唯一的实例，后续使用只涉及到了“读取”这个操作；
而在“懒汉模式”下，就会存在一些线程安全问题，我们接下来仔细分析此处可能存在的线程安全问题：

1. 由“随机调度”引起的线程安全问题
   因为在“懒汉模式”下，既涉及到了读的操作，也涉及到了写的操作。如果在多线程中出现这样的执行顺序，就会创建出多个实例，在我们要求“单例模式”的初心下，就出现了问题。
   
   这样的问题，也很容易解决，无非是通过加锁将读写操作打包到一起，使其成为“原子性”的操作。

public static Singleton getInstance() {synchronized (locker) {if (instance == null) {instance = new Singleton();}}return instance;}

这样的话，instance == null 和 instance=new Singleton()这两个操作就被打包到一起，就不会出现这样的问题了。但是，这份代码仍有一些问题。

2. 效率上的问题
   上述线程安全问题，其实只会在第一次创建实例的时候有可能会出现。也就是在整个使用过程中，只有第一次需要加锁，而我们现在的策略是每次读取instance实例都要加锁，就会造成效率低下。

public static Singleton getInstance() {if(instance==null) {synchronized (locker) {if (instance == null) {instance = new Singleton();}}}return instance;}

对于这样的问题，我们可以在外层再加一层判定，只在第一次创建实例的时候才加锁。

这样的代码，第一次见可能还是会有些疑惑，明明是两个一样的判定条件，怎么就需要两次？
首先我们分析，外面这个判定不要了可以吗？不行，这会影响效率。
那么，里面这个判定不要了可以吗？也不可以，如果里面这层判定没有，本质上instance == null 和 instance=new Singleton()这两步操作没有被打包到一起，仍会存在线程安全的问题。

那么，两层判定都不能去掉，究竟有何玄机呢？
这其实是因为加锁会引起阻塞，也就是意味着这两个判定执行的时间是不相同的，在阻塞的这段时间内，别的线程可能完成了一些操作（比如此处的新建实例），使得判定不再成立。

解决了效率上的问题，还可能会存在一些问题，我们继续分析。

3. 由“系统优化”引起的指令重排序问题

instance = new Singleton();

这条代码，可简单理解为由三个步骤组成：
（1）申请一段内存空间；
（2）在这个内存上调用构造方法，创建出这个实例；
（3）把这个内存地址赋值给Instance。

正常情况下，系统是按1、2、3的步骤执行的，也有可能会优化为1、3、2的执行顺序。如果在单线程中，1、3、2的执行顺序不会有什么问题，但在多线程中，未必如此。

如果是这样的执行顺序，那t2线程返回的instance实例，就是一个未初始化的“全0”的值，就会出现问题。

那么这样的问题，如何解决？
依靠于Java中提供的volatile关键字，我们之前提到它可以强制系统从内存中读取数据、避免出现“内存可见性问题”。此处是volatile的第二个功能，禁止指令重排序。

private volatile static Singleton instance=null;

到此，这段代码就实现了既高效，且避免了线程安全的问题。完整代码如下：

class Singleton{private volatile static Singleton instance=null;private static Object locker;public static Singleton getInstance() {if(instance==null) {synchronized (locker) {if (instance == null) {instance = new Singleton();}}}return instance;}//构造方法私有，在类外部不允许new新的实例private Singleton(){}}

二、“阻塞队列”的实现

2.1阻塞队列

阻塞队列，是基于普通队列做出的扩展（先进先出），优势在于：

1. 线程安全
2. 具有阻塞特性
   （1）如果一个队列已经满了，此时入队列就会阻塞，直到队列不满；
   （2）如果一个队列已经空了，此时出队列就会阻塞，直到队列不空。

在 Java 标准库中内置了阻塞队列. 如果我们需要在⼀些程序中使⽤阻塞队列, 直接使⽤标准库中的即可.

• BlockingQueue 是⼀个接⼝. 由ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue实现。
• put ⽅法⽤于阻塞式的⼊队列, take ⽤于阻塞式的出队列.
• BlockingQueue 也有 offer, poll, peek 等⽅法, 但是这些⽅法不带有阻塞特性.

2.2生产者消费者模型

引入生产者消费者模型，

1. 一方面是为了“解耦合”，也就是说比如原来是生产者和消费者直接交互，那么如果消费者变更、或者生产者变更，我们的代码就需要有比较大的改动，现在统一面向缓冲区（阻塞队列），就可以很好的“解耦合”。
2. 另一方面是“削峰填谷”，通过这个阻塞队列可以很好的控制消费者读取信息的速率。避免短时间内出现大量数据，引起问题。

2.3 阻塞队列的实现

实现阻塞队列，实质上就是实现三个方面：

1. 先实现普通队列

class MyBlockingQueue{private String[] elems=null;private int head=0;private int tail=0;private int size=0;public MyBlockingQueue(int capacity){elems=new String[capacity];}public void put(String elem){if(size >= elems.length){//队列满了，此处要能够阻塞}elems[tail]=elem;tail++;if(tail>= elems.length){tail=0;}size++;}public String take(){String elem=null;if(size == 0){//队列空了，要能阻塞}elem=elems[head];head++;if(head >= elems.length){head=0;}size--;return elem;}}

2. 再加上线程安全
   我们要求入队列和出队列、多次入队列、多次出队列不能并行执行，就要通过加锁实现。那锁都要包含那些操作呢？
   
   那么这个判断队列是否满了的操作是否需要包裹？
   
   试想这样的执行顺序，如果是入队列最后一个元素，就会出现问题。所以加锁应该包含判断队列是否为满的操作。

    public void put(String elem) {synchronized (locker) {if (size >= elems.length) {//队列满了，此处要能够阻塞}elems[tail] = elem;tail++;if (tail >= elems.length) {tail = 0;}size++;}}public String take() {synchronized (locker) {String elem = null;if (size == 0) {//如果队列为空，实现阻塞}elem = elems[head];head++;if (head >= elems.length) {head = 0;}size--;return elem;}}

3. 再加上阻塞功能
   我们通过wait()和notify()实现。

public void put(String elem) throws InterruptedException {synchronized (locker) {if (size >= elems.length) {//队列满了，此处要能够阻塞locker.wait();}elems[tail] = elem;tail++;if (tail >= elems.length) {tail = 0;}size++;locker.notify();}}public String take() throws InterruptedException {synchronized (locker) {String elem = null;if (size == 0) {locker.wait();}elem = elems[head];head++;if (head >= elems.length) {head = 0;}size--;locker.notify();return elem;}}

看到这样的代码，想必大家心中会有一个问题，你这一个锁里又wait()、又notify()的，这能行吗？不是自己把自己唤醒了吗？
但实际上这种问题还真不会出现。

真实情况应该是这样：

虽然我们刚才所说的这种问题不会出现，但仍有可能出现“连环唤醒”问题：

这里我们假设队列空了进入阻塞，我们通过put()加入一个元素后，这里随机唤醒一个线程进行take()操作，这个线程执行过程中，又把另一个线程唤醒进行take(),此时就会出现问题。

如何解决？
关键在于此处的 if 只判定一次。第一个线程被唤醒取走一个元素、并将第二个线程唤醒后，第二个线程如果能在判定1次队列是否为空，此时就不会出现问题。

这也很容易实现，我们只需要将 if 改成 while 即可。

2.4 再谈生产者消费者模型

我们可以通过通过控制入队或者出队的速度，来模拟这一过程。

//生产者
Thread t1=new Thread(()->{int n=1;while(true){try {queue.put(n+"");System.out.println("生产元素 "+n);n++;//2. 生产慢消费快Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});
//消费者Thread t2=new Thread(()->{while(true){try {String n=queue.take();System.out.println("消费元素 "+n);//1. 生产块消费慢Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});

1. 生产快消费慢
   

2. 生产慢消费快
   

完整代码：

class MyBlockingQueue{private String[] elems=null;private int head=0;private int tail=0;private int size=0;private static Object locker=new Object();public MyBlockingQueue(int capacity){elems=new String[capacity];}public void put(String elem) throws InterruptedException {synchronized (locker) {while(size >= elems.length) {//队列满了，此处要能够阻塞locker.wait();}elems[tail] = elem;tail++;if (tail >= elems.length) {tail = 0;}size++;locker.notify();}}public String take() throws InterruptedException {synchronized (locker) {String elem = null;while(size == 0) {locker.wait();}elem = elems[head];head++;if (head >= elems.length) {head = 0;}size--;locker.notify();return elem;}}}public class ThreadDemo4 {public static void main(String[] args) {MyBlockingQueue queue=new MyBlockingQueue(5);//生产者Thread t1=new Thread(()->{try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}int n=1;while(true){try {queue.put(n+"");System.out.println("生产元素 "+n);n++;//Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});//消费者Thread t2=new Thread(()->{while(true){try {String n=queue.take();System.out.println("消费元素 "+n);Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}
}