并发总结和实战 - 02 转账过程死锁及应对方案 - 《Java 并发专题笔记》

1. 死锁产生
2. 如何避免死锁
3. 死锁总结
4. 基于 Guarded Suspension 模式，优化百万交易系统

1. 死锁产生

1.1. 问题描述

在互联网的交易系统中，出现这种场景：【账户A】转账给【账户B】，同时【账户B】也转账给【账户A】，两个账户都需要锁住余额，所以通常会申请两把锁。转账时，先锁住自己的账户，并获取对方的锁，保证同一时刻只能由一个线程去执行转账。

这时可能就会出现，对方给我转账，同时我也给对方转账，那么双方都持有自己的锁，且尝试去获取对方的锁，这就造成可能一直申请不到对方的锁，循环等待发生『死锁』问题

1.2. 死锁示例

死锁是两个或两个以上的线程在执行过程中，互相持有对方所需要的资源，导致这些线程处于等待状态，无法继续执行。

public class DeadLock {
    public static String obj1 = "obj1";
    public static String obj2 = "obj2";
    public static void main(String[] args) {
        Thread a = new Thread(new Lock1());
        Thread b = new Thread(new Lock2());
        a.start();
        b.start();
    }
}
class Lock1 implements Runnable {
    @Override
    public void run() {
        try {
            System.out.println("Lock1 running");
            while (true) {
                synchronized (DeadLock.obj1) {
                    System.out.println("Lock1 lock obj1");
                    Thread.sleep(3000);
                    synchronized (DeadLock.obj2) {
                        System.out.println("Lock1 lock obj2");
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
class Lock2 implements Runnable {
    @Override
    public void run() {
        try {
            System.out.println("Lock2 running");
            while (true) {
                synchronized (DeadLock.obj2) {
                    System.out.println("Lock2 lock obj2");
                    Thread.sleep(3000);
                    synchronized (DeadLock.obj1) {
                        System.out.println("Lock2 lock obj1");
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

1.3. 死锁产生的原因

虽然进程在运行过程中，可能发生死锁，但死锁的发生也必须具备一定的条件，死锁的发生必须具备以下四个必要条件：

互斥，共享资源 X 和 Y 只能被一个线程占用；
占有且等待，线程01 已经取得共享资源 X，在等待共享资源 Y 的时候，不释放共享资源 X；
不可抢占，其他线程不能强行抢占线程01 占有的资源；
循环等待，线程01 等待线程02 占有的资源，线程02 等待线程01 占有的资源，就是循环等待。

2. 如何避免死锁

首先，“互斥”是没有办法避免的，对于“占用且等待”这个条件，我们可以一次性申请所有的资源，这样就不存在等待了。
对于“不可抢占”这个条件，占用部分资源的线程进一步申请其他资源时，如果申请不到，可以在一定时间后，主动释放它占有的资源，这样就解决了不可抢占这个条件。
对于“循环等待”，我们可以靠按“次序”申请资源来预防。所谓按序申请，就是给资源设定顺序，申请的时候可以先申请序号小的资源，再申请序号大的，这样资源线性化后，自然就不存在循环等待了。

2.1 破坏占用且等待条件

我们要破坏占用且等待，就是一次性申请占有所有的资源，我们拿文章开头的【账户A】、【账户B】来举例，就是一次性申请账户A，账户B的锁，当线程 01 拿到账户 A、B 全部的锁后，再执行具体的操作。

Allocator.java

/**
 * 分配器
 */
public class Allocator {
    private List<Object> als = new ArrayList<>();
    /**
     * 一次性申请所有资源
     *
     * @param from
     * @param to
     * @return
     */
    synchronized boolean apply(Object from, Object to) {
        if (als.contains(from) || als.contains(to)) {
            return false;
        } else {
            als.add(from);
            als.add(to);
        }
        return true;
    }
    synchronized void clean(Object from, Object to) {
        als.remove(from);
        als.remove(to);
    }
    private void Allocator() {
    }
    private static class SingleTonHolder {
        private static Allocator INSTANCE = new Allocator();
    }
    public static Allocator getInstance() {
        return SingleTonHolder.INSTANCE;
    }
}

Account.java

/**
 * 账户
 */
public class Account {
    private Allocator actr = Allocator.getInstance();
    private int balance;
    /**
     * 转账
     *
     * @param target 目标账户
     * @param amt    转账金额
     */
    void transfer(Account target, int amt) {
        while (!actr.apply(this, target)) {
        }
        try {
            synchronized (this) {
                System.out.println(this.toString() + " lock obj1");
                synchronized (target) {
                    System.out.println(this.toString() + " lock obj2");
                    if (this.balance > amt) {
                        this.balance -= amt;
                        target.balance += amt;
                    }
                }
            }
        } finally {
            //执行完后，再释放持有的资源
            actr.clean(this, target);
        }
    }
}

main

public class DeadLock {
    public static void main(String[] args) {
        Account a = new Account();
        Account b = new Account();
        a.transfer(b, 100);
        b.transfer(a, 200);
    }
}

2.2 破坏不可抢占条件

破坏不抢占条件，需要发生死锁的线程能够主动释放它占有的资源，但使用 synchronized 是做不到的。原因为 synchronized 申请不到资源时，线程直接进入了阻塞状态，而线程进入了阻塞状态也就没有办法释放它占有的资源了。

不过 JDK 中的 java.util.concurrent 提供了 Lock 解决这个问题。

显式使用 Lock 类中的定时 tryLock 功能来代替内置锁机制，可以检测死锁和从死锁中恢复过来。使用内置锁的线程获取不到锁会被阻塞，而显式锁可以指定一个超时时限（Timeout），在等待超过该时间后 tryLock 就会返回一个失败信息，也会释放其拥有的资源。

public class DeadLock {
    public static ReentrantLock lock1 = new ReentrantLock();
    public static ReentrantLock lock2 = new ReentrantLock();
    public static void main(String[] args) {
        Thread a = new Thread(new Lock1());
        Thread b = new Thread(new Lock2());
        a.start();
        b.start();
    }
    static class Lock1 implements Runnable {
        @Override
        public void run() {
            try {
                System.out.println("Lock1 running");
                while (true) {
                    if (lock1.tryLock(1, TimeUnit.MILLISECONDS)) {
                        System.out.println("Lock1 lock obj1");
                        //Thread.sleep(3000);
                        if (lock2.tryLock(1, TimeUnit.MILLISECONDS)) {
                            System.out.println("Lock1 lock obj2");
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock1.unlock();
                lock2.unlock();
            }
        }
    }
}

2.3 破坏循环等待条件

破坏这个条件，只需要对系统中的资源进行统一编号，进程可在任何时刻提出资源申请，必须按照资源的编号顺序提出。这样做就能保证系统不出现死锁。这就是『资源有序分配法』。

class Account {
    private int id;
    private int balance;
    void transfer(Account target, int amt){
        Account left = this;
        Account right = target;
        if (this.id > target.id) {
            left = target;
            right = this;
        }
        synchronized(left){
            synchronized(right){
                if (this.balance > amt){
                    this.balance -= amt;
                    target.balance += amt;
                }
            }
        }
    }
}

3. 死锁总结

4. 基于 Guarded Suspension 模式，优化百万交易系统

4.1 Guarded Suspension 模式简介

guarded 在这里是“保护”的意思；suspension 在这里是“暂时挂起”的意思。所以，Guarded Suspension 模式又称为“保护性暂挂模式”；

在多线程开发中，常常为了提高应用程序的并发性，会将一个任务分解为多个子任务交给多个线程并行执行，而多个线程之间相互协作时，仍然会存在一个线程需要等待另外的线程完成后继续下一步操作。而 Guarded Suspension 模式可以帮助我们解决上述的等待问题。

还是用交易系统的“转账”场景来讲述这个模式的实现。在上一篇文章中，我们提到，【账户A】转账给【账户B】，线程01需要持有账户A的锁，同时也需要持有账户B的锁，如果线程01拿不到两个锁，则进行 while(!actr.apply(this, target)) 死循环的方式来循环等待，直到一次全部获取到两个锁后，才进行后面的转账操作。

在并发量不大的情况下，这种方案还是可以接受的，但是一旦并发量增大，获取锁的冲突增加的时候，这种方案就不适合了，因为在这种场景下，可能要循环上万次才能获得锁，非常消耗性能，互联网高并发下显然不适合。

在这种场景下，最好的方案就是使用 Guarded Suspension 模式，如果线程 01 拿不到所有的锁，就阻塞自己，进入“等待WAITING”状态。当线程 01 要求的所有条件都满足后，“通知”等待状态的线程 01 重新执行。

4.2 代码举例

下面我们写一段代码来描述这段“等待-通知”机制：

02 转账过程死锁及应对方案 - 图3

1、创建GuardedQueue类

public class GuardedQueue {
    private final Queue<Integer> sourceList;
    public GuardedQueue() {
        this.sourceList = new LinkedBlockingQueue<>();
    }
    public synchronized Integer get() {
        while (sourceList.isEmpty()) {
            try {
                wait();    // <--- 如果队列为null，等待            
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return sourceList.peek();
    }
    public synchronized void put(Integer e) {
        sourceList.add(e);
        notifyAll();  //<--- 通知，继续执行    }
}

2、测试一下

public class App {
    public static void main(String[] args) {
        GuardedQueue guardedQueue = new GuardedQueue();
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        executorService.execute(() -> {
                    guardedQueue.get();
                }
        );
        Thread.sleep(2000);
        executorService.execute(() -> {
                    guardedQueue.put(20);
                }
        );
        executorService.shutdown();
        executorService.awaitTermination(30, TimeUnit.SECONDS);
    }
}

4.3 总结与拓展

Guarded Suspension模式的“等待-通知”机制是一种非常普遍的线程间协作的方式。我们在平时工作中经常看到有同学使用“轮询 while(true)”的方式来等待某个状态，其实都可以用这个“等待-通知”机制来优化。

另外，有同学可能会问为什么不用 **notify()** 来实现通知机制呢？
Notify() 和 notifyAll() 这两者是有区别的，notify() 是会随机地通知等待队列中的任意一个线程，而 notifyAll() 会通知等待队列中的所有线程。

觉得 notify() 会更好一些的同学可能认为即便通知所有线程，也只有一个线程能够进入临界区。但是实际上使用 notify() 也很有风险，因为随机通知等待的线程，可能会导致某些线程永远不会被通知到。

所以除非经过深思熟虑，否则尽量使用 notifyAll()。