Java 并发设计模式

时间 2019-11-08

标签 java 并发设计模式栏目 Java 繁體版

原文原文链接

Java 并发设计模式

1、Thread Local Storage 模式

1. ThreadLocal 的使用

Thread Local Storage 表示线程本地存储模式。java

大多数并发问题都是因为变量的共享致使的，多个线程同时读写同一变量便会出现原子性，可见性等问题。局部变量是线程安全的，本质上也是因为各个线程各自拥有本身的变量，避免了变量的共享。设计模式

Java 中使用了 ThreadLocal 来实现避免变量共享的方案。ThreadLocal 保证在线程访问变量时，会建立一个这个变量的副本，这样每一个线程都有本身的变量值，没有共享，从而避免了线程不安全的问题。缓存

下面是 ThreadLocal 的一个简单使用示例：安全

public class ThreadLocalTest {

    private static final ThreadLocal<SimpleDateFormat> threadLocal =
            ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    public static SimpleDateFormat safeDateFormat() {
        return threadLocal.get();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask<SimpleDateFormat> task1 = new FutureTask<>(ThreadLocalTest::safeDateFormat);
        FutureTask<SimpleDateFormat> task2 = new FutureTask<>(ThreadLocalTest::safeDateFormat);

        Thread t1 = new Thread(task1);
        Thread t2 = new Thread(task2);
        t1.start();
        t2.start();
        System.out.println(task1.get() == task2.get());//返回false，表示两个对象不相等
    }
}

程序中构造了一个线程安全的 SimpleDateFormat ，两个线程取到的是不一样的示例对象，这样就保证了线程安全。多线程

2. ThreadLocal 原理浅析

线程 Thread 类内部有两个 ThreadLocalMap 类型的变量：并发

/* ThreadLocal values pertaining to this thread. This map is maintained
 * by the ThreadLocal class. */
ThreadLocal.ThreadLocalMap threadLocals = null;

/*
 * InheritableThreadLocal values pertaining to this thread. This map is
 * maintained by the InheritableThreadLocal class.
 */
ThreadLocal.ThreadLocalMap inheritableThreadLocals = null;

其中第二个变量的用途是建立可继承父线程变量的子线程，只不过这并不经常使用，主要介绍第一个。异步

ThreadLocalMap 是一个用于存储 ThreadLocal 的特殊 HashMap，map 中 key 就是 ThreadLocal，value 是线程变量值。只不过这个 map 并不被 ThreadLocal 持有，而是被 Thread 持有。工具

当调用 ThreadLocal 类中的 set 方法时，就会建立 Thread 中的 threadLocals 属性。性能

//ThreadLocal的set方法
public void set(T value) {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);//获取Thread中的ThreadLocalMap
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
}

能够看到，最终的 ThreadLocal 对象和变量值并非建立在 ThreadLocal 内部，而是 Thread 中的 ThreadLocalMap，ThreadLocal 在这里只是充当了代理的做用。this

3. ThreadLocal 内存泄漏问题

存储数据的 TheadLocalMap 被 Thread 持有，而不是 ThreadLocal，主要的缘由即是 ThreadLocal 的生命周期比 Thread 要长，若是 ThreadLocal 对象一直存在，那么 map 中的线程就不能被回收，容易致使内存泄漏。

而 Thread 持有 ThreadLocalMap，而且 ThreadLocalMap 对 ThreadLocal 的引用仍是弱引用，这样当线程被回收时，map 也可以被回收，更加安全。

可是 Java 的这种设计并无彻底避免内存泄漏问题。若是线程池中的线程存活时间过长，那么其持有的 ThreadLocalMap 一直不会被释放。ThreadLocalMap 中的 Entry 对其 value 是强引用的（对 ThreadLocal 是弱引用），这样就算 ThreadLocalMap 的生命周期结束了，可是 value 值并无被回收。

解决的办法即是手动释放 ThreadLocalMap 中对 value 的强引用，可使用 TheadLocal 的 remove 方法。在 finally 语句块中执行。例以下面这个简单的示例：

public class ThreadLocalTest {

    private final ThreadLocal<Integer> threadLocal = new ThreadLocal<>();

    public void test(){
        //设置变量值
        threadLocal.set(10);
        try {
            System.out.println(threadLocal.get());
        }
        finally {
            //释放
            threadLocal.remove();
        }
    }
}

2、Immutability 模式

1. 不可变的概念

Immutability，即不变模式。能够理解为只要对象一经建立，其状态是不可以被改变的，没法进行写操做。

要实现 Immuatability 模式很简单，将一个类自己及其全部的属性都设为 final ，而且方法都是只读的，须要注意的是，若是类的属性也是引用类型，那么其对应的类也要知足不可变的特性。final 应该都很熟悉了，用它来修饰类和方法，分别表示类不可继承、属性不可改变。

Java 中具有不可变性的类型包括：

String
final 修饰的基本数据类型
Integer、Long、Double 等基本数据类型的包装类
Collections 中的不可变集合

具有不可变性的类，若是须要有相似修改这样的功能，那么它不会像普通的对象同样改变本身的属性，而是建立新的对象。

下面是 String 的字符串链接方法 concat() 的源码，仔细观察，能够看到最后方法返回的时候，建立了一个新的 Sring 对象：

public String concat(String str) {
    int otherLen = str.length();
    if (otherLen == 0) {
        return this;
    }
    int len = value.length;
    char buf[] = Arrays.copyOf(value, len + otherLen);
    str.getChars(buf, len);
    //建立新的对象
    return new String(buf, true);
}

而 Collections 工具能够将集合变为不可变的，彻底禁止写、修改等操做。示例以下：

//Collections 中构建不可变集合的方法
Collections.unmodifiableList();
Collections.unmodifiableSet();
Collections.unmodifiableMap();
Collections.unmodifiableSortedSet();
Collections.unmodifiableSortedMap();
---
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
//构建不可变集合
List<Integer> unmodifiableList = Collections.unmodifiableList(list);

unmodifiableList.remove(1);//抛出异常

2. 对象池

对于一个不可变性的类，若是频繁的对其进行修改操做，那么一直会建立性新的对象，这样就比较浪费内存空间了，一种解决办法即是利用对象池。

原理也很简单，新建对象的时候，去对象池看是否存在对象，若是存在则直接利用，若是不存在才会建立新的对象，建立以后再将对象放到对象池中。

以长整型的包装类 Long 为例，它缓存了 -128 到 127 的数据，若是建立的是这个区间的对象，那么会直接使用缓存中的对象。例如 Long 中的 valueOf 方法就用到了这个缓存，而后直接返回：

public static Long valueOf(long l) {
    final int offset = 128;
    //在这个区间则直接使用缓存中的对象
    if (l >= -128 && l <= 127) { // will cache
        return LongCache.cache[(int)l + offset];
    }
    return new Long(l);
}

3、Guarded Suspension 模式

1. Guarded Suspension 实现

Guarded Suspension 意为保护性暂停。一个典型的使用场景是：当客户端线程 T 发送请求后，服务端这时有大量的请求须要处理，这时候就须要排队，线程 T 进入等待状态，直到服务端处理完请求而且返回结果。

Guarded Suspension 的实现很简单，有一个对象 GuardedObject，其内部有一个属性，即被保护的对象，还有两个方法，客户端调用 get() 方法，若是未获取到结果，则进入等待状态，即“保护性暂停”；还有一个 notice() 通知方法，当服务端处理完请求后，调用这个方法，而且唤醒等待中的线程。示意图以下：

示例代码以下：

public class GuardedObject<T> {

    private T obj;
    private final Lock lock = new ReentrantLock();
    private final Condition finished = lock.newCondition();

    //调用方线程获取结果
    T get(){
        lock.lock();
        try {
            while (未获取到结果){
                finished.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        finally {
            lock.unlock();
        }
        return obj;
    }

    //执行完后通知
    void notice(T obj){
        lock.lock();
        try {
            this.obj = obj;
            finished.signalAll();
        }
        finally {
            lock.unlock();
        }
    }
}

从代码中能够看到，Guarded Suspension 模式本质上就是一种等待-通知机制，只不过使用这种模式，在解决实际的问题的时候，须要根据状况进行程序功能的扩展。

2. 使用示例

仍是上面提到的那个例子，当客户端发送请求后，须要等待服务端的响应结果，这时候就可使用 Guarded Suspension 来实现，下面是代码示例：

public class SendRequest<T> {

    //至关于消息队列
    private final BlockingQueue<Request> queue = new ArrayBlockingQueue<>(5);

    //客户端发送请求
    void send(Request request) throws InterruptedException {
        //将消息存放至队列中
        queue.put(request);
        //建立Guarded Suspension模式的对象
        GuardedObject<Request> guardedObject = GuardedObject.create(request.id);
        //循环等待，获取结果
        Request res = guardedObject.get(Objects::nonNull);
    }

    //服务端处理请求
    void handle() throws InterruptedException {
        //从队列中获取请求
        Request request = queue.take();
        //调用请求对应的GuardedObject，并处理请求
        GuardedObject.handleRequest(request.id, request);
    }
    
    //请求类
    private static class Request{
        private int id;
        private String content;
    }
}

须要注意的是，这里并非直接使用 new GuardedObject() 的方式来建立对象，这是由于须要找到每一个请求和对象之间的对应关系，因此 GuardedObject 内部使用了一个 map 来保存对象，key 是对应的请求 id。

GuardedObject 类代码以下：

public class GuardedObject<T> {

    private T obj;

    private final Lock lock = new ReentrantLock();
    private final Condition finished = lock.newCondition();

    private static final ConcurrentHashMap<Integer, GuardedObject> map = new ConcurrentHashMap<>();

    //建立对象
    public static GuardedObject create(int id){
        GuardedObject guardedObject = new GuardedObject();
        //保存对象和请求的对应关系
        map.put(id, guardedObject);
        return guardedObject;
    }

    //处理请求
    public static void handleRequest(int id, Object obj){
        GuardedObject guardedObject = map.remove(id);
        if (guardedObject != null){
            
            //具体的处理逻辑省略
            
            //处理完后通知
            guardedObject.notice(obj);
        }
    }

    //调用方线程获取结果
    T get(Predicate<T> p){
        lock.lock();
        try {
            while (!p.test(obj)){
                finished.await();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        finally {
            lock.unlock();
        }
        return obj;
    }

    //执行完后通知
    void notice(T obj){
        lock.lock();
        try {
            this.obj = obj;
            finished.signalAll();
        }
        finally {
            lock.unlock();
        }
    }
}

4、Balking 模式

Balking 模式的典型应用场景是，业务逻辑依赖于某个条件变量的状态，所以这种模式又能够理解为多线程版本的 if。

public class BalkingTest {

    private boolean flag = false;

    public void execute(){
        if (!flag){
            return;
        }
        //具体的执行操做省略
        flag = false;
    }

    public void test(){
        //省略业务代码若干
        flag = true;
    }
}

例如上面这个例子，一段业务逻辑会改变 flag 的值，另外一个方法会根据 flag 的值来决定是否继续执行。

这个程序并非线程安全的，解决的办法也很简单，就是加互斥锁，而后能够将改变 flag 值的逻辑单独拿出来，以下：

public class BalkingTest {
    private boolean flag = false;

    public synchronized void execute(){
        if (!flag){
            return;
        }
        //具体的执行操做省略
        flag = false;
    }

    public void test(){
        //省略业务代码若干
        change();
    }
    
    public synchronized void change(){
        flag = true;
    }
}

Balking 模式通常可使用互斥锁来实现，而且能够将对条件变量的改变的逻辑和业务逻辑进行分离，这样可以减少锁的粒度，提高性能。Balking 模式大多应用于须要快速失败的场景，即当条件变量不知足，则直接失败。这也是它和 Guarded Suspension 模式的区别，由于 Guarded Suspension 模式在条件不知足的时候，会一直等待条件知足。

5、Worker - Thread 模式

Worker Thread 模式，对应到现实世界，相似工厂中的工人作任务，当有任务的时候，工人取出任务执行。

解决的办法是使用线程池，而且使用一个阻塞队列来存储任务，线程池中的线程从队列中取出任务执行。线程池的使用须要注意几点：

任务队列尽可能使用有界队列，避免任务过多形成 OOM。
应该明确指定拒绝策略，能够根据实际状况实现 RejectedExecutionHandler 接口自定义拒绝策略。
应该给线程指定一个有意义的名字，最好和业务相关。
为不一样的任务建立不一样的线程池，这样可以有效的避免死锁问题。

6、Two - Phase Termination 模式

1. 两阶段终止概念

Two - Phase Termination，即两阶段终止，主要是为解决如何正确的终止一个线程，这里说的是一个线程终止另外一个线程，而不是线程终止本身。

Java 中的线程提供了一个 stop() 方法用来终止线程，这不过这个方法会直接将线程杀死，风险过高，而且这个方法已经被标记为废弃，不建议使用了。

两阶段终止，即将线程的结束分为了两个阶段，第一个阶段是一个线程 T1 向另外一个线程 T2 发送终止指令，第二个阶段是线程 T2 响应终止指令。

根据 Java 的线程状态，线程若是要进入 TERMINATED 状态则必须先进入 RUNNABLE 状态，而处于 RUNNABLE 状态的线程有可能转换到休眠状态。

Java 的线程提供了 interrupt() 方法，这个方法的做用即是将线程的状态从休眠状态转换到 RUNNABLE 状态。

切换到 RUNNABLE 状态以后，线程有两种方式能够终止，一是执行完 run() 方法，自动进入终止状态；二是设置一个标志，线程若是检测到这个标志，则退出 run() 方法，这就是两阶段终止的响应终止指令。

2. 程序示例

下面是一个简单的使用 interrupt() 方法和中断标志位来终止线程的示例：

public class Test {
    public static void main(String[] args) {

        Thread thread = new Thread(() -> {
            //检测到中断则退出
            while (!Thread.currentThread().isInterrupted()){
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    //从新设置中断标志
                    Thread.currentThread().interrupt();
                }
                System.out.println("I am roseduan");
            }
        });
        thread.start();
        thread.interrupt();
    }
}

程序要每隔三秒打印语句，可是线程启动以后就直接调用了 interrupt() 方法，因此线程直接退出了。须要注意的是这里在捕获异常以后，须要从新设置线程的中断状态，由于 JVM 的异常处理会清除线程的中断状态。

在实际的生产中，并不推荐使用这种方式，由于在 Thread 内部可能会调用其余的方法，而其余的方法并不可以保证正确的处理了线程中断，解决的办法即是自定义一个线程的中断标志，以下所示：

public class Test {
    //自定义中断标志
    private volatile boolean isTerminated = false;
    private Thread thread;

    public synchronized void start(){
        thread = new Thread(() -> {
            //检测到中断则退出
            while (!isTerminated) {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    //从新设置中断状态
                    Thread.currentThread().interrupt();
                }
                System.out.println("I am roseduan");
            }
            isTerminated = false;
        });
        thread.start();
    }

    //线程终止方法
    public synchronized void stop(){
        isTerminated = true;
        thread.interrupt();
    }
}

3. 终止线程池

Java 中并不太会显式的建立和终止一个线程，使用更多的是线程池。

Java 中的线程池提供了两个方法来终止，分别是 shutdown() 和 shutdownNow() ，两个方法的区别以下：

shutdown()：拒绝新的任务，等待正在执行的和已经在阻塞队列中的任务执行完后，再关闭线程池
shutdownNow()：直接关闭线程池，拒绝新的任务，而且中断正在执行的任务，已经在阻塞队列中的任务也不会被执行了。

7、Producer - Consumer 模式

这是较为经常使用的生产者 - 消费者模式，Java 中的线程池就使用了这种模式，线程的使用方是生产者，提供任务，线程池自己是消费者，取出并执行任务。

生产者 - 消费者模式使用了一个任务队列，生产者将任务添加到队列中，消费者从队列中取出任务执行。

这样的设计的目的有三个：

解耦，生产者和消费者之间没有直接的关联，而是经过队列进行通讯。
其次能够实现异步，例如生产者能够不用管消费者的行为，直接将任务添加到队列中。消费者也能够不在意生产者，直接从队列中取任务。
最后，能够平衡生产者和消费者之间的速度差别。

下面是一个简单的生产者 - 消费者程序示例：

public class ProducerConsumerTest {

    private BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100);

    public void produce() {
        queue.add(new Task());

    }

    public void consume() {
        Task task = queue.poll();
        while (task != null){
            task.execute();
            task = queue.poll();
        }
        System.out.println("没有任务了");
    }

    public static void main(String[] args) throws InterruptedException {
        Test test = new Test();
        //生产者线程，建立10个任务
        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                test.produce();
            }
        });
        producer.start();
        producer.join();

        //消费者线程
        Thread consumer = new Thread(test::consume);
        consumer.start();

    }
}

class Task{
    public void execute(){
        System.out.println("执行任务");
    }
}