并发编程中的设计模式

在并发编程的过程中，对于某些特定的问题，一般都有特定的解决方案来处理，就好像是设计模式一样，它们具有通用的模板。可以把这些解决方案称为并发编程中的设计模式。

这篇文章结合《Java源码世界》专题中并发工具类以及多线程的相关文章进行总结，形成的一种并发编程的技巧。

一、线程终止

在多线程开发的时候，经常会用到，在某个线程中，需要去终止另外一个线程，Thread类也提供了终止线程的方法stop()，但在JDK中该方法被标记为@Deprecated，原因就在于该方法做过于暴力。stop()会直接真正杀死线程，如果这时被杀死的线程锁住了共享资源，当它被杀死后就再也没有机会释放锁了，其他线程就会永远无法获取锁。

使用System.exit(int)同样也可以停止线程，但这会导致整个程序都停止。

而优雅的停止线程的方式就是要借助线程的中断机制，Java线程的中断方法interrupt()并不会直接就去中断线程，它只是去设置了线程的中断状态，当线程运行到某个阶段的时候，它会去检查中断状态，进而进行中断。这样就不会出现锁无法释放的情况了。

Java中断机制整个过程分为两阶段(Two-phase Termination,两阶段终止)，第一个阶段线程T1向线程T2发送终止指令(保证被中断的线程要处于RUNNABLE状态)，而第二个阶段是线程T2响应终止指令。

Java线程进入终止状态的前提是线程进入RUNNABLE状态，因为它要保证线程中的run()方法的逻辑被执行完了。而利用Java中断机制的interrupt()方法，可以让线程从休眠状态转换成RUNNABLE状态，再从RUNNABLE状态转换到终止状态。优雅的方式就是让Java线程执行完run()方法，我们通常自定义设置一个中断标志，然后线程在合适的时机检查自定义的中断标志，如果发现符合终止条件，就自动退出run()方法。

这种两阶段提交的线程终止的方式是一个很常用的方式，但在Java中两阶段终止需要注意两个关键点：

仅检查自定义的中断标志是不够，因为线程的状态可能处于休眠状态(需要将其唤醒)
仅检查线程的中断状态也是不够的，因为如果我们依赖别的库，它可能没有很好的处理中断异常，比如在捕获到Thread.sleep()抛出的异常后，没有重新设置线程的中断状态，那么就会导致线程不能正常终止。

注：可以自定义线程的标志位用于终止线程

以下面的程序为例，如果当调用interrupt()方法后，run()里面的Thread.sleep(2000)就会抛出中断异常，同时会清除中断标志位，如果没有在catch代码块中重新设置线程的中断状态，循环就没法退出了。

//采集线程
Thread rptThread;

//启动采集功能
synchronized void start() {
    rptThread = new Thread(() -> {
        while (!Thread.currentThread().isInterrupted()) {
            //省略采集、回传实现
            report();
            //每隔两秒钟采集、回传一次数据
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                //重新设置线程中断状态
                // Thread.currentThread().interrupt();
            }
        }
    });
    rptThread.start();
}

private void report() {
    System.out.println("采集数据");
}

//终止采集功能
synchronized void stop() {
    rptThread.interrupt();
}


public static void main(String[] args) {
    MonitorProxy monitor = new MonitorProxy();
    monitor.start();

    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    monitor.stop();
}

在上面的基础上，加上自定义的中断标志位，就可以保证线程一定会退出循环，在stop()中将中断标志置为true，这样就算线程的中断状态为false，线程依然会退出循环

//采集线程
Thread rptThread;

//线程终止标志位
volatile boolean terminated = false;

//启动采集功能
synchronized void start() {
    rptThread = new Thread(() -> {

        while (!Thread.currentThread().isInterrupted()&&!terminated) {
            //省略采集、回传实现
            report();
            //每隔两秒钟采集、回传一次数据
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                //重新设置线程中断状态
                //Thread.currentThread().interrupt();
            }
        }
    });
    rptThread.start();
}

private void report() {
    System.out.println("采集数据");
}

//终止采集功能
synchronized void stop() {
    //设置中断标志位
    terminated = true;
    rptThread.interrupt();
}

使用场景：

安全地终止线程，比如释放该释放的资源；
确保终止处理逻辑在线程结束之前一定会执行时，可使用该方法；

二、避免共享

多线程中共享资源的写操作造就了线程安全问题，如果能够避免对共享资源的写，或者只对写进行加锁，甚至将共享资源变成线程私有的，通过这些操作来避免共享。

Immutability模式，Copy-on-Write模式，Thread-Specific Storage模式本质上都是为了避免共享。但在使用这几种模式时需要注意一些问题：

使用Immutability模式时，如果类的属性为实例对象，不仅要保证实例对象的不可变性，更好保证实例对象的属性的不可变性
Copy-on-Write模式需要注意拷贝的性能问题
Thread-Specific Storage需要注意异步执行的问题

2.1 Immutability

“多个线程同时读写同一共享变量存在并发问题”这句话的必要条件之一就是读写共存，如果只有读，没有写，并不会出现什么并发问题了。解决并发问题就简单的方法就是让共享变量只有读操作而没有写操作，这就是不变性模式的根本。

简单来说，就是当对象被创建完成之后，它的所有属性都将是不可更改的，对象的状态不会再发生任何变化。

实现方式就是通过final关键字，将一个类所有的属性都设置为final的，并且只允许有只读方法，这样一个类就具有了不可变性。更严格的做法是将类也加上final，保证其不能被继承，也就不能随意扩展。

JDK中有很多类都具有不可变性，例如经常用到的String以及Integer、Long、Double等基本类型的包装类都具有不可变性，这些对象的安全性都是靠不可变性来保证的。它们都遵守不可变类的三点要求：类和属性都是不可变的，所有方法都是只读的。

但是在使用Immutability的时候，需要特别注意一下两点：

对象的所有属性都是不可变的，并不能保证不可变性
不可变对象也需要正确发布(不可变对象作为属性值时，没有final修饰也不能保证不可变性)

下面通过两个例子来演示不变性的容易出现出错的地方。使用的时候一定确认保持不变性的边界在哪里，是否要求属性对象也具备不可变性。

下面的代码中，Order的属性User虽然是final的，依然可以通过setAge()方法来设置age的属性age。

class User{
    int age=24;
    int name="lizhi";
}
final class Order {
    final User user;
    void setAge(int a){
        user.age=a;
    }
}

虽然不可变对象是线程安全，但并不意味着引用这些不可变对象的对象也是线程安全的

下面的代码中，虽然User类及属性都是线程安全的，但它作为其他类的引用对象时，并没有用final修饰，就导致了对这个user的修改在多线程中并不能保证可见性和原子性。

final class User{
    final int age=24;
    final String name="lizhi";
}

class Order {
    User user;
    void setFoo(User user){
        this.user=user;
    }
}

2.2 Copy-on-Write

Java的String类型在实现replace()和concat()方法时，并不会更改原字符串里面value[]数组的内容，而是创建一个新字符串，这种方法在解决不可变对象的修改问题时经常用到。但这种解决方案本质上就是一种Copy-on-Write的形式，可以缩写为COW或CoW，顾名思义就是写时复制。

Copy-on-Write的应用非常广泛，它是最简单的解决并发问题的方案。Java中的引用数据类型Sting、Integer、Long等都是基于Copy-on-Write方案实现的。

Copy-on-Write最大的缺点就是消耗内存，每次修改都需要复制一个新的对象出来，但随着JDK版本的升级，GC的性能也越来越好了，这种内存消耗也是可以接受的。在实际工作中，如果写操作非常少(读多写少)，就可以考虑使用Copy-on-Write。

应用场景

在Java中，CopyOnWriteArrayList和CopyOnWriteArraySet这两个Copy-on-Write容器，其背后的设计思路都是Copy-on-Write；通过Copy-on-Write前面这两个容器的读操作都是无锁的，正因为无锁，读操作的性能发挥到了极致。只需要对写操作进行加锁，添加元素的时候直接获取独占锁，然后重新创建个数组，将数据复制过去即可。

public boolean add(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
        Object[] newElements = Arrays.copyOf(elements, len + 1);
        newElements[len] = e;
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

Copy-on-Write在操作系统中应用也很广泛，类Unix的操作系统中创建进程的API是fork()，传统的fork()函数会创建父进程的一个完整副本，例如父进程的地址空间现在用到了1G的内存，那么fork()子进程的时候要复制父进程整个进程的地址空间（占有 1G 内存）给子进程，这个过程是很耗时的。而Linux中fork()子进程的时候，并不复制整个进程的地址空间，而是让父子进程共享同一个地址空间；只用在父进程或者子进程需要写入的时候才会复制地址空间，从而使父子进程拥有各自的地址空间。

Copy-on-Write应用最多的地方还是函数式编程领域。函数时编程的基础是不可变性，所以函数式编程里面的所有修改操作都是通过Copy-on-Write来解决的。

有一些RPC框架的注册中心，通过Copy-on-Write设计思想来维护路由表。路由表典型的是读多写少的，而且路由表对数据的一致性要求并不高，服务动态上线时，反馈到客户端的路由表里面，即使有10s的延迟也没什么太大影响。

2.3 Thread-Specific Storage

Thread-Specific Storage(线程本地存储)模式，即使数据操作只有一个入口，其内部会为每个线程分配私有的存储空间。在Java标准库中，ThreadLocal就实现了这种模式，Thread类里面有一个ThreadLocal.ThreadLocalMap，就是为每个线程分配数据存储空间。其底层是一个用一个数组来存储，根据每个ThreadLocal对象计算出当前该ThreadLocal对象中该线程存储的数组的索引值，核心代码如下：

private Entry getEntry(ThreadLocal<?> key) {
    int i = key.threadLocalHashCode & (table.length - 1);
    Entry e = table[i];
    if (e != null && e.get() == key)
        return e;
    else
        return getEntryAfterMiss(key, i, e);
}

线程本地存储模式本质上是一种避免共享的方案，由于没有共享，所以自然也就没有并发问题。如果需要在并发场景中使用一个线程不安全的工具类，最简单的方案就是避免共享。避免共享有两种方案，一种方案是将这个工具类作为局部变量使用，另外一种方案就是线程本地存储模式。

局部变量方案的缺点是在高并发场景下会频繁创建对象，而线程本地存储方案，每个线程只需要创建一个工具类的实例，所以不存在频繁创建对象的问题。

应用场景：

SimpleDateFormat不是线程安全的，那如果需要在并发场景下使用它，有一个办法就是用ThreadLocal来解决。

static class SafeDateFormat {
    //定义ThreadLocal变量
    static final ThreadLocal<DateFormat> tl=ThreadLocal.withInitial(
        ()-> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    static DateFormat get(){
        return tl.get();
    }
}

注意：在线程池中使用ThreadLocal需要避免内存泄漏和线程安全的问题

ExecutorService es;
ThreadLocal tl;
es.execute(()->{
  //ThreadLocal增加变量
  tl.set(obj);
  try {
    // 省略业务逻辑代码
  }finally {
    //手动清理ThreadLocal 
    tl.remove();
  }
});

三、多线程判断(if)

3.1 Guarded Suspension

在多线程线程中，可能存在一种情况，线程A的执行依赖于线程B的计算结果，那么线程B在执行前就需要判断结果是否已经计算完成，如果结果还没有计算完成，线程B就应该被阻塞。当线程A完成计算之后，再去唤醒因等待计算结果而阻塞的线程，这种称之为Guarded Suspension(保护性暂停模式)。

Guarded Suspension模式是通过让线程等待来保护实例的安全性，即守护-挂起模式。

Guarded Suspension模式允许多个线程对实例资源进行访问，但是实例资源需要对资源的分配做出管理。

Guarded Suspension模式也常被称作Guarded Wait模式、Spin Lock模式（因为使用了 while 循环去等待），它还有一个更形象的非官方名字：多线程版本的 if。

这种模式通常有以下使用场景：

某个结果需要从线程A传递到线程B，需要这两个线程关联同一个GuardedObject
如果有结果不断从线程A传递到线程B，则可以使用消息队列来完成
JDK中，join()和Future的实现，采用的就是此模式

因为要等待另一方的结果，因此归类到同步模式。等待唤醒机制的规范实现。此模式依赖于Java线程的阻塞唤醒机制：

sychronized+wait/notify/notifyAll
reentrantLock+Condition(await/singal/singalAll)
cas+park/unpark

阻塞唤醒机制底层原理：linux pthread_mutex_lock/unlock pthread_cond_wait/singal

解决线程之间的协作不可避免会用到阻塞唤醒机制

Guarded Suspension模式的实现：

public class GuardedObject<T> {
    //结果
    private T obj;
    //获取结果
    public T get(){
        synchronized (this){
            //没有结果等待   防止虚假唤醒
            while (obj==null){
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return obj;
        }
    }
    //产生结果
    public void complete(T obj){
        synchronized (this){
            //获取到结果，给obj赋值
            this.obj = obj;
            //唤醒等待结果的线程
            this.notifyAll();
        }
    }
}

3.2 Balking

Balking是“退缩不前”的意思。如果此时已经不再适合执行这个操作，或者没必要执行这个操作，就停止处理，直接返回。当流程的执行顺序依赖于某个共享变量的场景，或者说某段代码只需要有一个线程执行成功就可以了，这种也可以归纳为多线程if模式。Balking 模式常用于一个线程发现另一个线程已经做了某一件相同的事，那么本线程就无需再做了，直接结束返回。

Balking模式是一种多个线程执行同一操作A时可以考虑的模式；在某一个线程B被阻塞或者执行其他操作时，其他线程同样可以完成操作A，而当线程B恢复执行或者要执行操作A时，因A已被执行，而无需线程B再执行，从而提高了B的执行效率。

Balking模式和Guarded Suspension模式一样，存在守护条件，如果守护条件不满足，则中断处理；这与Guarded Suspension模式不同，Guarded Suspension模式在守护条件不满足的时候会一直等待至可以运行。

常见应用场景：

sychronized轻量级锁膨胀过程中，只需要一个线程膨胀获取monitor对象即可
dcl, Double Checked Locking，典型的就是创建单例对象时
服务组件初始化，只需要执行依次初始化逻辑即可

实现Blaking的方式有如下几种：

锁机制(synchronized reentrantLock)
CAS
对于共享变量不要求原子性的场景，可以使用volatile

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

boolean changed=false;
// 自动存盘操作
void autoSave(){
    synchronized(this){
        if (!changed) {
            return;
        }
        changed = false;
    }
    // 执行存盘操作
    // 省略且实现
    this.execSave();
}
// 编辑操作
void edit(){
    // 省略编辑逻辑
    ......
    change();
}
// 改变状态
void change(){
    synchronized(this){
        changed = true;
    }
}

还有常见的单次初始化

boolean inited = false;
synchronized void init(){
    if(inited){
        return;
    }
    //省略doInit的实现
    doInit();
    inited=true;
}

四、多线程分工

这个模式是比较容易理解的，每个线程干自己的活就可以了，典型的多线程分工模式有三种Thread-Per-Message、Worker Thread和生产者-消费者模式

Thread-Per-Message模式需要注意线程的创建，销毁以及是否会导致OOM。
Worker Thread模式需要注意死锁问题，提交的任务之间不要有依赖性。
生产者 – 消费者模式可以直接使用线程池来实现

4.1 Thread-Per-Message

Thread-Per-Message模式是最简单实用的分工方法，就是为每个任务分配一个独立的线程。但如果任务量很多的时候，可能会导致OOM的问题。所以任务很多事这样方式就不适用了。

Thread-Per-Message模式的一个最经典的应用场景是早期的网络编程里服务端的实现，服务端为每个客户端请求创建一个独立的线程，当线程处理完请求后，自动销毁，这是一种最简单的并发处理网络请求的方法。

final ServerSocketChannel  ssc= ServerSocketChannel.open().bind(new InetSocketAddress(8080));
//处理请求    
try {
    while (true) {
        // 接收请求
        SocketChannel sc = ssc.accept();
        // 每个请求都创建一个线程
        new Thread(()->{
            try {
                // 读Socket
                ByteBuffer rb = ByteBuffer.allocateDirect(1024);
                sc.read(rb);
                //模拟处理请求
                Thread.sleep(2000);
                // 写Socket
                ByteBuffer wb = (ByteBuffer)rb.flip();
                sc.write(wb);
                // 关闭Socket
                sc.close();
            }catch(Exception e){
                throw new UncheckedIOException(e);
            }
        }).start();
    }
} finally {
    ssc.close();
}

Thread-Per-Message模式作为一种最简单的分工方案，Java中使用会存在性能缺陷。在 Java 中的线程是一个重量级的对象，创建成本很高，一方面创建线程比较耗时，另一方面线程占用的内存也比较大。所以为每个请求创建一个新的线程并不适合高并发场景。为了解决这个缺点，Java 并发包里提供了线程池等工具类。
在其他编程语言里，例如Go语言，基于轻量级线程实现Thread-Per-Message模式就完全没有问题。
对于一些并发度没那么高的异步场景，例如定时任务，采用Thread-Per-Message模式是完全没有问题的。

4.2 Worker Thread

要想有效避免线程的频繁创建、销毁以及 OOM 问题，就不得不提 Java 领域使用最多的Worker Thread模式。Worker Thread模式可以类比现实世界里车间的工作模式：车间里的工人，有活儿了，大家一起干，没活儿了就聊聊天等着。Worker Thread模式中Worker Thread对应到现实世界里，其实指的就是车间里的工人。

上面服务端的例子，采用线程池来实现：

ExecutorService es = Executors.newFixedThreadPool(200);
final ServerSocketChannel ssc = ServerSocketChannel.open().bind(new InetSocketAddress(8080));
//处理请求    
try {
    while (true) {
        // 接收请求
        SocketChannel sc = ssc.accept();
        // 将请求处理任务提交给线程池
        es.execute(()->{
            try {
                // 读Socket
                ByteBuffer rb = ByteBuffer.allocateDirect(1024);
                sc.read(rb);
                //模拟处理请求
                Thread.sleep(2000);
                // 写Socket
                ByteBuffer wb = 
                    (ByteBuffer)rb.flip();
                sc.write(wb);
                // 关闭Socket
                sc.close();
            }catch(Exception e){
                throw new UncheckedIOException(e);
            }
        });
    }
} finally {
    ssc.close();
    es.shutdown();
}

Worker Thread模式能避免线程频繁创建、销毁的问题，而且能够限制线程的最大数量。Java 语言里可以直接使用线程池来实现Worker Thread模式，线程池是一个非常基础和优秀的工具类，甚至有些大厂的编码规范都不允许用new Thread()来创建线程，必须使用线程池。

4.3 生产者-消费者模式

生产者 – 消费者模式的核心是一个任务队列，生产者线程生产任务，并将任务添加到任务队列中，而消费者线程从任务队列中获取任务并执行。

public class Test {
    public static void main(String[] args) {
        // 生产者线程池
        ExecutorService producerThreads = Executors.newFixedThreadPool(3);
        // 消费者线程池
        ExecutorService consumerThreads = Executors.newFixedThreadPool(2);
        // 任务队列，长度为10
        ArrayBlockingQueue<Task> taskQueue = new ArrayBlockingQueue<Task>(10);
        // 生产者提交任务
        producerThreads.submit(() -> {
            try {
                taskQueue.put(new Task("任务"));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        // 消费者处理任务
        consumerThreads.submit(() -> {
            try {
                Task task = taskQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    static class Task {
        // 任务名称
        private String taskName;
        public Task(String taskName) {
            this.taskName = taskName;
        }
    }
}

队列在生产者-消费者模式的优点：

异步处理

场景：用户注册后，需要发注册邮件和注册短信。传统的做法有两种 1.串行的方式；2.并行方式

引入消息队列，将不是必须的业务逻辑异步处理

解耦

场景：用户下单后，订单系统需要通知库存系统扣减库存。

平衡生产者生产与消费者速度

在计算机当中，创建的线程越多，CPU进行上下文切换的成本就越大，所以我们在编程的时候创建的线程并不是越多越好，而是适量即可，采用生产者和消费者模式就可以很好的支持我们使用适量的线程来完成任务。

如果在某一段业务高峰期的时间里生产者“生产”任务的速率很快，而消费者“消费”任务速率很慢，由于中间的任务队列的存在，也可以起到缓冲的作用，我们在使用MQ中间件的时候，经常说的削峰填谷也就是这个意思。

过饱问题解决方案

在实际生产项目中会有些极端的情况，导致生产者/消费者模式可能出现过饱的问题。单位时间内，生产者生产的速度大于消费者消费的速度，导致任务不断堆积到阻塞队列中，队列堆满只是时间问题。
思考：是不是只要保证消费者的消费速度一直比生产者生产速度快就可以解决过饱问题？

我们只要在业务可以容忍的最长响应时间内，把堆积的任务处理完，那就不算过饱。

什么是业务容忍的最长响应时间？
比如埋点数据统计前一天的数据生成报表，第二天老板要看的，你前一天的数据第二天还没处理完，那就不行，这样的系统我们就要保证，消费者在24小时内的消费能力要比生产者高才行。

场景一：消费者每天能处理的量比生产者生产的少；如生产者每天1万条，消费者每天只能消费5千条。
解决办法：消费者加机器
原因：生产者没法限流，因为要一天内处理完，只能消费者加机器

场景二：消费者每天能处理的量比生产者生产的多。系统高峰期生产者速度太快，把队列塞爆了
解决办法：适当的加大队列
原因：消费者一天的消费能力已经高于生产者，那说明一天之内肯定能处理完，保证高峰期别把队列塞满就好

场景三：消费者每天能处理的量比生产者生产的多。条件有限或其他原因，队列没法设置特别大。系统高峰期生产者速度太快，把队列塞爆了
解决办法：生产者限流
原因：消费者一天的消费能力高于生产者，说明一天内能处理完，队列又太小，那只能限流生产者，让高峰期塞队列的速度慢点

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/153607.html