协作对象死锁及其解决方案

协作对象死锁及其解决方案

1.前言

在遇到转账等的需要保证线程安全的情况时，我们通常会使用加锁的方式来保证线程安全，但如果无法合理的使用锁，很可能导致死锁。或者有时我们使用线程池来进行资源的使用，如调用数据库，但无法合理使用锁也可能导致死锁。
Java程序无法自动检测死锁并预防，在Java程序中如果遇到死锁将会是一个非常严重的问题，轻则导致线程阻塞，程序响应时间变长，系统吞吐量变小；重则导致系统部分功能失去响应能力无法使用。因此我们应当预防和规避这些问题。

如果对Synchronized关键字的基本使用方法不是很清楚的话可以看一下这篇文章：
Java中Synchronized关键字的基本使用方法

2.死锁说明

死锁是指两个或多个线程在同一资源上相互占用，并请求锁定对方的资源，从而导致恶性循环的现象。
死锁产生的四个必要条件：
1）互斥条件：资源是独占且排他的，即任意时刻一个资源只能给一个线程使用。其他线程若申请一个资源，而该资源被另一线程占用时，则申请者只能等待，直到资源被占用者释放。
2）不可剥夺条件：线程所获得的资源在未使用完毕之前，不会被其他线程强行剥夺，而只能由获得该资源的线程进行释放。
3）请求和保持条件：线程每次申请它所需要的一部分资源，在申请新的资源的同时，继续占用已分配到的资源。
4）循环等待条件：在发生死锁时必然存在一个线程等待队列{P1,P2,…,Pn}，其中P1等待P2占有的资源，P2等待P3占有的资源，…，Pn等待P1占有的资源，形成一个线程等待环路，环路中每一个线程所占有的资源同时被另一个申请。

协作对象死锁

死锁的产生往往不像顺序死锁那样明显（关于顺序死锁可以看一下这本篇文章：顺序死锁及其解决方案），就算其存在死锁风险往往也只会在高并发的场景下才可能暴露出来（这并不意味着应用没有高并发就不用考虑死锁问题了，作为开发者，你无法预测用户的操作）。
接下来介绍一种隐藏的比较深的死锁，这种死锁往往产生于多个协作对象的函数调用不透明。

Coordinate：坐标类，记录出租车的经度和纬度。
Fleet： 出租车车队类，车队类包含两个集合：车队中所有车辆信息taxis和车队中当前空闲的出租车信息available，此外还提供获取车队中所有出租车当前地址信息快照的方法getImage()
Taxi：出租车类，出租车属于某个出租车车队Fleet，此外还包含当前坐标location和目的地坐标destination，出租车在更新目的地信息的时候会判断当前坐标与目的地坐标是否相等，相等则会通知所属车队车辆空闲，可以接收下一个目的地
Image： 车辆地址信息快照类，用于获取出租车的地址信息

/*** 坐标类*/
public class Coordinate {/*** 经度*/private Double longitude;/*** 纬度*/private Double latitude;public Coordinate(Double longitude, Double latitude) {this.longitude = longitude;this.latitude = latitude;}public Double getLongitude() {return longitude;}public void setLongitude(Double longitude) {this.longitude = longitude;}public Double getLatitude() {return latitude;}public void setLatitude(Double latitude) {this.latitude = latitude;}@Overridepublic String toString() {return "Coordinate{" +"longitude=" + longitude +", latitude=" + latitude +'}';}}

/*** 车队类 -> 调度管理出租车*/
public class Fleet {/*** 车队中所有出租车*/private final Set<Taxi> taxis;/*** 车队中目前空闲的出租车*/private final Set<Taxi> available;public Fleet(Set<Taxi> taxis) {this.taxis = this.available = taxis;}/*** 出租车到达目的地后调用该方法，向车队发出当前出租车空闲信息** @param taxi*/public synchronized void free(Taxi taxi) {System.out.println("出租车到站了");available.add(taxi);}/*** 获取所有出租车在不同时刻的地址快照*需要获取当前车队Fleet的锁，以及在遍历出租车获取其地址信息时需要获取每个出租车Taxi对象的锁* @return*/public synchronized Image getImage() {Image image = new Image();for (Taxi taxi : taxis) {image.drawMarker(taxi);}return image;}}

/*** 出租车类*/
public class Taxi {/*** 出租车唯一标志*/private String id;/*** 当前坐标*/private Coordinate location;/*** 目的地坐标*/private Coordinate destination;/*** 所属车队*/private final Fleet fleet;/*** 获取当前地址信息** @return*/public synchronized Coordinate getLocation() {return location;}/*** 更新当前地址信息* 如果当前地址与目的地地址一致，则表名到达目的地需要通知车队，当前出租车空闲可用前往下一个目的地*需要获取当前出租车Taxi对象的锁以及出租车所属车队Fleet的锁* @param location*/public synchronized void setLocation(Coordinate location) {this.location = location;if (location.equals(destination)) {fleet.free(this);}}public Coordinate getDestination() {return destination;}/*** 设置目的地** @param destination*/public synchronized void setDestination(Coordinate destination) {this.destination = destination;}public Taxi(Fleet fleet) {this.fleet = fleet;}public String getId() {return id;}public void setId(String id) {this.id = id;}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}Taxi taxi = (Taxi) o;return Objects.equals(location, taxi.location) &&Objects.equals(destination, taxi.destination);}@Overridepublic int hashCode() {return Objects.hash(location, destination);}
}

/*** 获取所有出租车在某一时刻的位置快照*/
public class Image {Map<String, Coordinate> locationSnapshot = new HashMap<>();public void drawMarker(Taxi taxi) {locationSnapshot.put(taxi.getId(), taxi.getLocation());System.out.println("出租车当前位置为：" + taxi.getLocation().toString());}}

public class TaxiDeadLock {public static void main(String[] args) {HashSet<Taxi> taxiList = new HashSet<Taxi>();Fleet fleet = new Fleet(taxiList);Taxi taxi1 = new Taxi(fleet);taxiList.add(taxi1);Coordinate coordinate = new Coordinate(1.1, 2.2);taxi1.setDestination(coordinate);taxi1.setLocation(new Coordinate(1.0, 2.0));for (int i = 0; i < 50; i++) {new Thread(() -> {fleet.getImage();}, "车队").start();new Thread(() -> {taxi1.setLocation(coordinate);}, "出租车").start();}}}

从图片上我们可以看出，出租车和车队互相锁住了对方所需要的资源，并且请求对方所占有的资源，进程进入了死锁状态。
让我们从代码层面上看：
车队调用getImage()方法获取所有出租车在不同时刻的地址快照时，会先锁住当前车队，之后遍历出租车集合获取每辆出租车地址信息，通过drawMarker(Taxi taxi)方法获取出租车地址信息，这时我们会调用taxi.getLocation()方法，此方法会锁住当前出租车。
出租车调用setLocation(Coordinate location)方法更新当前地址信息时，会先锁住当前出租车，之后当前地址与目的地地址进行判断，如果一致，则调用free(Taxi taxi)方法通知车队当前出租车空闲，此方法会锁住出租车所属的车队。
以上代码并非显式的在一个方法中对多个资源进行加锁，而是隐式的在多个类中对对多个方法进行加锁，粗看之下似乎没什么问题，但是细细分析，就会发现存在很大的死锁隐患。

使用同步方法可能会发生较长时间的阻塞，这就可能给了死锁的机会。基于此类问题，我们通常要对需求进行分析，然后尝试采用缩小锁的范围或者不加锁等方式来解决。

下面让我们分析一下上文代码中的死锁：车队获取所有出租车在不同时刻的地址快照时，是一定要加锁的，因为这并不是一个原子性操作，不加锁就有可能发生车队中原本是5辆车，循环的时候突然多了或者少了车的情况。出租车更新当前地址信息时同理。所以这两处的锁，我们是不可以去掉的。那我们只可以尝试采用缩小锁的范围来防范：

/*** 车队类 -> 调度管理出租车*/
public class Fleet {/*** 车队中所有出租车*/private final Set<Taxi> taxis;/*** 车队中目前空闲的出租车*/private final Set<Taxi> available;public Fleet(Set<Taxi> taxis) {this.taxis = this.available = taxis;}/*** 出租车到达目的地后调用该方法，向车队发出当前出租车空闲信息** @param taxi*/public synchronized void free(Taxi taxi) {System.out.println("出租车到站了");available.add(taxi);}/*** 优化内容* getImage()不再是同步方法* 将同步范围（锁住的代码）缩小* this（出租车车队对象）与drawMarker()方法中获取taxi对象的锁不再嵌套不会死锁** @return*/public Image getImage() {Set<Taxi> copy;synchronized (this) {copy = new HashSet<Taxi>(taxis);}Image image = new Image();for (Taxi taxi : copy) {image.drawMarker(taxi);}return image;}}

/*** 出租车类*/
public class Taxi {/*** 出租车唯一标志*/private String id;/*** 当前坐标*/private Coordinate location;/*** 目的地坐标*/private Coordinate destination;/*** 所属车队*/private final Fleet fleet;/*** 获取当前地址信息** @return*/public synchronized Coordinate getLocation() {return location;}/*** 优化内容* setLocation(Coordinate location)方法不在加锁* 将同步范围（锁住的代码）缩小* this的锁与fleet顺序获取 ，锁内没有嵌套，不会死锁* @param*/public void setLocation(Coordinate location) {this.location = location;boolean release = false;synchronized (this) {if (location.equals(destination)) {release = true;}}if (release) {fleet.free(this);}}public Coordinate getDestination() {return destination;}/*** 设置目的地** @param destination*/public synchronized void setDestination(Coordinate destination) {this.destination = destination;}public Taxi(Fleet fleet) {this.fleet = fleet;}public String getId() {return id;}public void setId(String id) {this.id = id;}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}Taxi taxi = (Taxi) o;return Objects.equals(location, taxi.location) &&Objects.equals(destination, taxi.destination);}@Overridepublic int hashCode() {return Objects.hash(location, destination);}
}

从上文代码中可以看出，我们优化了车队类的getImage()方法，将其同步方法改为了同步代码块，缩小了锁的范围：当调用getImage()方法时，我们创建一个车队拷贝，然后上锁，确保车队拷贝和当前车队一致，之后解锁，用车队拷贝去循环获取出租车的锁。之前会出现死锁是因为车队类既锁住了自己又去请求出租车的锁，现在请求出租车的时候没有锁住自己，破坏了死锁的四个触发条件之一，构不成死锁，出租车类的方法同理。
简单的说车队类和出租车类的方法我们只要修改一个就破坏了死锁的条件，无法构成死锁，但是我们还是需要对其不合理的代码进行优化。

由此我们可以得出结论：对于协作对象死锁，我们通常要对需求进行分析，然后尝试采用缩小锁的范围或者不加锁等方式来破坏死锁的四个必须条件中的一个或多个来解决。