Singleton 模式

单例模式是一种对象创建模式，用于产生一个对象的具体实例，它可以确保系统中一个类只产生一个实例。在 Java 中，这样的行为能带来两个好处：

  • 对于频繁使用的对象，可以省略 new 操作花费的时间，这对于那些重量级对象而言，是非常可观的一笔系统开销。
  • 由于 new 操作的次数减少，因而对系统内存的使用频率也会降低，这将减轻 GC 压力。

严格来说，单例模式与并行没有直接的关系，讨论这个模式，是因为这个模式太常见了，我们不可避免地会在多线程环境中使用它们，因此我们需要一种高效的单例实现。

静态内部类实现了延迟加载，同时保证了线程安全（依赖 JVM 来保证线程安全）不建议使用双重锁校验模式。

  1. public class StaticSingleton {
  2.     private StaticSingleton () {}
  4.     private static class SingletonHolder {
  5.         private static StaticSingleton instance = new StaticSingleton ();
  6.     }
  8.     public static StaticSingleton getinstance () {
  9.         return SingletonHolder.instance;
  10.     }
  11. }
  • 构造函数必须是 private 的，以控制不能随便创建这个类的实例。
  • instance 必须是 private 并且 static 的，否则 instance 的安全性无法得到保证。

    Immutability 模式

    多个线程同时读写同一共享变量存在并发问题，这里的必要条件之一是读写，如果只有读而没有写，是没有并发问题的。因此解决并发问题，最简单的办法就是让共享变量只有读操作，而没有写操作。这个办法如此重要，以至于被上升到了一种解决并发问题的设计模式：不变性（Immutability）模式。

所谓不变性，就是对象一旦被创建之后，状态就不再发生变化。即变量一旦被赋值，就不允许再被修改了，没有修改操作，也就保持了不变性。将一个类所有的属性都设置成 final 的，并且只有只读方法，那么这个类基本上就具备不可变性了。更严格的做法是这个类本身也是 final 的，也就是不允许继承。因为子类可以覆盖父类的方法，有可能改变不可变性。

不变模式的主要使用场景需满足以下两个条件：

  • 当对象创建后，其内部状态和数据不再发生任何变化
  • 对象需要被共享，被多线程频繁访问。

Java SDK 里很多类都具备不可变性，例如经常用到的 String、Integer、Double 等基础类型的包装类都具备不可变性，这些对象的线程安全性都是靠不可变性来保证的。如果你仔细翻看这些类的声明、属性和方法，你会发现它们都严格遵守不可变类的三点要求：类和属性都是 final 的，所有方法均是只读的。

如果具备不可变性的类，需要提供类似修改的功能，通常做法是创建一个新的不可变对象，这是与可变对象的一个重要区别，可变对象往往是修改自己的属性。但所有的修改操作都创建一个新的不可变对象，那是不是创建的对象太多了，有点太浪费内存呢？这样做的确有些浪费，那如何解决呢？

利用享元模式避免创建重复对象

利用享元模式（Flyweight Pattern）可以减少创建对象的数量，从而减少内存占用。Java 里的 Long、Integer、Short 等基本数据类型的包装类都用到了享元模式。

享元模式本质上其实就是一个对象池，利用享元模式创建对象的逻辑也很简单：创建之前，先去对象池里看看是不是存在；如果存在就利用对象池里的对象；如果不存在就会新创建一个对象，并且把这个新创建出来的对象放进对象池里。

下面以 Long 这个类为例，看它是如何利用享元模式来优化对象的创建的。Long 内部维护了一个静态的对象池，缓存了 [-128,127] 之间的数字，这个对象池在 JVM 启动时就创建好了，而且这个对象池一直都不会变化，也就是说它是静态的。之所以采用这样的设计，是因为 [-128,127] 之间的数字利用率最高。

  1. public static Long valueOf(long l) {
  2.     final int offset = 128;
  3.     if (l >= -128 && l <= 127) {
  4.         return LongCache.cache[(int)l + offset];
  5.     }
  6.     return new Long(l);
  7. }
  8. // [-128,127]直接的数字做了缓存
  9. private static class LongCache {
  10.     private LongCache(){}
  11.     static final Long cache[] = new Long[-(-128) + 127 + 1];
  12.     static {
  13.         for(int i = 0; i < cache.length; i++)
  14.             cache[i] = new Long(i - 128);
  15.     }
  16. }

Guarded Suspension 模式

想象如下场景：外出聚餐前我们预订了一个包间，到饭店后大堂经理看了一眼包间，发现服务员正在收拾，就会告诉我们预订的包间正在收拾，等包间收拾完后，大堂经理再带我们去包间就餐。在这个场景中，我们是否要等待，完全是由大堂经理来协调的。

在并发编程的角度，可以类比有很多桌客人，而大堂经理只有一个，通过这一个大堂经理来协调多个包间，等包间收拾完成这个条件满足后，大堂经理就会主动通知客人去对应包间就餐。这种设计模式被称为：Guarded Suspension，Guarded 是被守护、被保护的意思，Suspension 是暂停的意思，如果执行现在的处理会造成问题，就让执行处理的线程进行等待，通过让线程等待来保证实例的安全性。

下面我们来看看 Guarded Suspension 模式是如何模拟大堂经理这个角色的。

  1. public class GuardedObject<T> {
  2.     //受保护的对象
  3.     T obj;
  4.     private final Lock lock = new ReentrantLock();
  5.     private final Condition done = lock.newCondition();
  6.     private final int timeout=2;
  7.     // 保存所有GuardedObject
  8.     private final static Map<Object, GuardedObject> gos = new ConcurrentHashMap<>();
  10.     // 静态方法创建GuardedObject
  11.     public static <K> GuardedObject create(K key) {
  12.         GuardedObject go = new GuardedObject();
  13.         gos.put(key, go);
  14.         return go;
  15.     }
  16.     public static <K, T> void fireEvent(K key, T obj) {
  17.         GuardedObject go = gos.remove(key);
  18.         if (go != null){
  19.             go.onChanged(obj);
  20.         }
  21.     }
  23.     // 获取受保护对象  
  24.     public T get(Predicate<T> p) {
  25.         lock.lock();
  26.         try {
  27.             // MESA管程推荐写法
  28.             while(!p.test(obj)){
  29.                 done.await(timeout, TimeUnit.SECONDS);
  30.             }
  31.         } catch(InterruptedException e) {
  32.             throw new RuntimeException(e);
  33.         } finally {
  34.             lock.unlock();
  35.         }
  36.         return obj;
  37.     }
  39.     // 事件通知方法
  40.     public void onChanged(T obj) {
  41.         lock.lock();
  42.         try {
  43.             this.obj = obj;
  44.             done.signalAll();
  45.         } finally {
  46.             lock.unlock();
  47.         }
  48.     }
  49. }

上面的示例代码中，GuardedObject 内部维护了一个 Map，Key 是包间 ID，Value 是 GuardedObject 对象实例，同时增加了静态方法 create() 和 fireEvent()；create() 方法用来创建一个 GuardedObject 对象实例，并根据 key 值将其加入到 Map 中，而 fireEvent() 方法则是模拟的大堂经理根据包间找就餐人的逻辑。协调逻辑本质上就是根据 await 和 signalAll 这样的等待-唤醒机制来进行的。

Balking 模式

我们提到可以用“多线程版本的 if”来理解 Guarded Suspension 模式，不同于单线程中的 if，这个“多线程版本的 if”是需要等待的，而且还很执着，必须要等到条件为真。但世界上不是所有场景都需要这么执着，有时候我们还需要快速放弃。如果现在不适合执行这个操作，或者没必要执行这个操作，就立即返回，这样的场景就可以使用 Balking 模式，这样能够提高程序的响应性。

需要快速放弃的一个最常见的例子是各种编辑器提供的自动保存功能。自动保存功能的实现逻辑一般都是隔一定时间自动执行存盘操作，存盘操作的前提是文件做过修改，如果文件没有执行过修改操作，就需要快速放弃存盘操作。代码示例如下所示：

  1. boolean changed=false;
  3. //自动存盘操作
  4. void autoSave(){
  5.   synchronized(this){
  6.     if (!changed) {
  7.       return;
  8.     }
  9.     changed = false;
  10.   }
  11.   //执行存盘操作
  12.   //省略且实现
  13.   this.execSave();
  14. }
  16. //编辑操作
  17. void edit(){
  18.   //省略编辑逻辑
  19.   ......
  20.   change();
  21. }
  23. //改变状态
  24. void change(){
  25.   synchronized(this){
  26.     changed = true;
  27.   }
  28. }

上面我们用 synchronized 实现了 Balking 模式，这种实现方式最为稳妥。不过在某些特定场景下，我们也可以使用 volatile 来实现，但使用 volatile 的前提是对原子性没有要求。

Balking 模式有一个非常典型的应用场景就是单次初始化，下面的示例代码是它的实现。这个实现方案中，我们将 init() 声明为一个同步方法，这样同一个时刻就只有一个线程能够执行 init() 方法；init() 方法在第一次执行完时会将 inited 设置为 true，这样后续执行 init() 方法的线程就不会再执行 doInit() 了。

  1. class InitTest{
  2.     boolean inited = false;
  3.     synchronized void init(){
  4.         if(inited){
  5.             return;
  6.         }
  7.         //省略doInit的实现
  8.         doInit();
  9.         inited=true;
  10.     }
  11. }

Producer-Consumer 模式

生产者线程负责提交用户请求，消费者线程则负责具体处理生产者提交的任务。生产者和消费者之间则通过共享内存缓冲区进行通信。共享内存缓冲区作为生产者和消费者间的通信桥梁，避免了生产者和消费者直接通信。同时，由于内存缓冲区的存在，允许生产者和消费者存在执行上的性能差异，确保系统正常运行。
image.png

1. wait、notify

  1. public class ProConDemo {
  2.     // 队列容量
  3.     private static final int CAPACITY = 5;
  4.     private static int i = 0;
  5.     /**
  6.      * 生产者
  7.      */
  8.     static class Producer implements Runnable {  
  9.         private Queue<Integer> queue;
  10.         private int maxLength;
  11.         public Producer(Queue<Integer> queue, int maxLength) {
  12.             this.queue = queue;
  13.             this.maxLength = maxLength;
  14.         }
  15.         @Override
  16.         public void run() {
  17.             while (true) {
  18.                 synchronized (queue) {
  19.                     try {
  20.                         // 在循环中调用wait，防止虚假唤醒
  21.                         while (maxLength == queue.size()) {
  22.                             System.out.println("Queue is full, Producer[" + Thread.currentThread().getName() + "] " +
  23.                                     "waiting for consumer to take something from queue.");
  24.                             queue.wait();
  25.                         }
  26.                     } catch (InterruptedException e) {}
  27.                     System.out.println("[" + Thread.currentThread().getName() + "] Producing value :" + i);
  28.                     queue.offer(i++);
  29.                     queue.notifyAll();
  30.                 }
  31.             }
  32.         }
  33.     }
  35.     /**
  36.      * 消费者
  37.      */
  38.     static class Consumer implements Runnable {
  39.         private Queue<Integer> queue;
  40.         public Consumer(Queue<Integer> queue) {this.queue = queue;}
  41.         @Override
  42.         public void run() {
  43.             while (true) {
  44.                 synchronized (queue) {
  45.                     try {
  46.                         // 防止虚假唤醒
  47.                         while (queue.isEmpty()) {
  48.                             System.out.println("Queue is empty, Consumer[" + Thread.currentThread().getName() + "] " +
  49.                                     "thread is waiting for Producer");
  50.                             queue.wait();
  51.                         }
  52.                     } catch (InterruptedException e) {}
  53.                     Integer element = queue.poll();
  54.                     System.out.println("[" + Thread.currentThread().getName() + "] Consuming value :" + element);
  55.                     queue.notifyAll();
  56.                 }
  57.             }
  58.         }
  59.     }
  60. }

2. await、signal

  1. public class ProConDemo {
  2.     private static final int CAPACITY = 5;
  3.     private static int i = 0;
  4.     private static Lock lock = new ReentrantLock();
  5.     private static Condition full = lock.newCondition();
  6.     private static Condition empty = lock.newCondition();
  7.     /**
  8.      * 生产者
  9.      */
  10.     static class Producer implements Runnable {
  11.         private Queue<Integer> queue;
  12.         private int maxLength;
  13.         public Producer(Queue<Integer> queue, int maxLength) {
  14.             this.queue = queue;
  15.             this.maxLength = maxLength;
  16.         }
  17.         @Override
  18.         public void run() {
  19.             while (true) {
  20.                 lock.lock();
  21.                 try {
  22.                     while (maxLength == queue.size()) {
  23.                         System.out.println("Queue is full, Producer[" + Thread.currentThread().getName() + "] " +
  24.                                 "waiting for consumer to take something from queue.");
  25.                         full.await();
  26.                     }
  27.                     System.out.println("[" + Thread.currentThread().getName() + "] Producing value :" + i);
  28.                     queue.offer(i++);
  29.                     // 唤醒其他所有消费者
  30.                     empty.signalAll();
  31.                 } catch (InterruptedException e) {
  32.                     e.printStackTrace();
  33.                 } finally {
  34.                     lock.unlock();
  35.                 }
  36.             }
  37.         }
  38.     }
  39.     /**
  40.      * 消费者
  41.      */
  42.     static class Consumer implements Runnable {
  43.         private Queue<Integer> queue;
  44.         public Consumer(Queue<Integer> queue) { this.queue = queue;}
  45.         @Override
  46.         public void run() {
  47.             while (true) {
  48.                 lock.lock();
  49.                 try {
  50.                     while (queue.isEmpty()) {
  51.                         System.out.println("Queue is empty, Consumer[" + Thread.currentThread().getName() + "] " +
  52.                                 "thread is waiting for Producer");
  53.                         empty.await();
  54.                     }
  55.                     Integer element = queue.poll();
  56.                     System.out.println("[" + Thread.currentThread().getName() + "] Consuming value :" + element);
  57.                     // 唤醒其他所有生产者
  58.                     full.signalAll();
  59.                 } catch (InterruptedException e) {
  60.                     e.printStackTrace();
  61.                 } finally {
  62.                     lock.unlock();
  63.                 }
  64.             }
  65.         }
  66.     }
  67. }

3. BlockingQueue

  1. public class ProConDemo {
  2.     // 阻塞队列长度
  3.     private static final int CAPACITY = 5;
  4.     private static volatile AtomicInteger count = new AtomicInteger(0);
  5.     /**
  6.      * 生产者
  7.      */
  8.     static class Producer implements Runnable {
  9.         private BlockingQueue<Integer> blockingQueue;
  10.         private volatile boolean isRunning = true;
  11.         public Producer(LinkedBlockingQueue<Integer> queue) {this.blockingQueue = queue;}
  12.         @Override
  13.         public void run() {
  14.             while (isRunning) {
  15.                 try {
  16.                     blockingQueue.put(count.incrementAndGet());
  17.                     System.out.println("[" + Thread.currentThread().getName() + "] Producing value :" + count.get());
  18.                 } catch (InterruptedException e) {
  19.                     e.printStackTrace();
  20.                 }
  21.             }
  22.         }
  23.         public void stop() {
  24.             isRunning = false;
  25.         }
  26.     }
  28.     /**
  29.      * 消费者
  30.      */
  31.     static class Consumer implements Runnable {
  32.         private BlockingQueue<Integer> blockingQueue;
  33.         public Consumer(LinkedBlockingQueue<Integer> queue) {this.blockingQueue = queue;}
  34.         @Override
  35.         public void run() {
  36.             while (true) {
  37.                 try {
  38.                     Integer x = blockingQueue.take();
  39.                     System.out.println("[" + Thread.currentThread().getName() + "] Consuming value :" + x);
  40.                 } catch (InterruptedException e) {
  41.                     e.printStackTrace();
  42.                 }
  43.             }
  44.         }
  45.     }
  46. }

4. Disruptor

Disruptor 是一款高性能的有界内存队列，对标的是 JDK 提供的 BlockingQueue 系列容器。详情链接

Future 模式

Future 模式的核心思想是异步调用。当我们需要调用一个函数时，如果这个函数执行得很慢，那么我们就要进行等待。但有时候，我们可能并不急着要结果。因此，我们可以让被调者立即返回，让它在后台慢慢处理这个请求。对于调用者来说，则可以先处理一些其他任务，在真正需要数据的场合再去尝试获得需要的数据。

Future 模式有点类似在网上买东西。如果我们在网上下单买了一部手机，当我们支付完成后，手机并没有办法立即送到家里，但是在电脑上会立即产生一个订单。这个订单就是将来发货或者领取手机的重要凭证，这个凭证也就是 Future 模式中会给出的一个契约。在支付活动结束后，大家不会傻傻地等着手机到来，而是各忙各的。而这张订单就成了商家发货的驱动力。当然，这一切我们不关心，我们只需要在快递上门时收货而已。

Future 模式很常用，因此 JDK 内部已经为我们准备好了一套完整的实现。首先，我们先看一下 Future 模式的基本结构，如下图所示：
FutureTask.png
其中 Future 接口类似于订单或者说是契约。通过它，你可以得到真实的数据。RunnableFuture 继承了 Future 和 Runnable 两个接口，其中 run() 方法用于构造真实的数据。它有一个具体的实现 FutureTask 类，FutureTask 类有一个内部类 Sync，一些实质性的工作会委托 Sync 类来实现。而 Sync 类最终会调用 Callable 接口，完成实际数据的组装工作。

Callable 接口只有一个方法 call()，它会返回需要构造的实际数据。这个 Callable 接口也是 Future 框架和应用程序之间的重要接口。要实现自己的业务系统，通常需要实现自己的 Callable 对象。此外，FutureTask 类也与应用密切相关，通常可以使用 Callable 实例构造个 FutureTask 实例，并将它提交给线程池。

Two-Phase Termination 模式

Java 语言的 Thread 类中曾经提供了一个 stop() 方法用来终止线程，可是早已不建议使用了，原因是这个方法用的就是一剑封喉的做法，被终止的线程没有机会料理后事。既然不建议使用 stop() 方法，那在 Java 领域，我们又该如何优雅地终止线程呢？

前辈们经过认真对比分析总结出了一套成熟的方案，叫做两阶段终止模式。顾名思义，就是将终止过程分成两个阶段，第一个阶段主要是线程 T1 向线程 T2 发送终止指令，而第二阶段则是线程 T2 响应终止指令。

  • 通过 Thread 类提供的 interrupt() 方法，将线程转换到 RUNNABLE 状态。

  • 终止的优雅方式是让 Java 线程自己执行完 run() 方法，所以一般我们会设置一个标志位，然后线程会在合适的时机检查这个标志位，如果发现符合终止条件，则自动退出 run() 方法。

综合上面这两点，我们能总结出终止指令，其实包括两方面内容：interrupt() 方法和线程终止的标志位。理解了两阶段终止模式之后，下面我们看一个实际工作中的案例。

  1. public class Proxy {
  2.     // 线程终止标志位
  3.     private volatile boolean terminated = false;
  4.     // 线程启动标志位，防止重复启动
  5.     private boolean started = false;
  6.     private Thread rptThread;
  8.     public synchronized void start(){
  9.         if (started) {
  10.             return;
  11.         }
  12.         started = true;
  13.         terminated = false;
  14.         rptThread = new Thread(()->{
  15.             while (!terminated){
  16.                 report();
  17.                 try {
  18.                     Thread.sleep(2000);
  19.                 } catch (InterruptedException e){
  20.                     //重新设置线程中断状态
  21.                     Thread.currentThread().interrupt();
  22.                 }
  23.             }
  24.             started = false;
  25.         });
  26.         rptThread.start();
  27.     }
  29.     public synchronized void stop(){
  30.         // 设置中断标志位
  31.         terminated = true;
  32.         // 中断线程rptThread
  33.         rptThread.interrupt();
  34.     }
  35. }

需要注意的是，我们在捕获 Thread.sleep() 的中断异常后，通过 Thread.currentThread().interrupt() 重新设置了线程的中断状态，因为 JVM 的异常处理会清除线程的中断状态。