Java高并发编程：活跃性危险

落日映苍穹つ 2022-03-28 09:39 207阅读 0赞

Java高并发程序中，不得不出现资源竞争以及一些其他严重的问题，比如死锁、线程饥饿、响应性问题和活锁问题。在安全性与活跃性之间通常存在依赖，我们使用加锁机制来确保线程安全，但是如果过度地使用加锁，则可能导致锁顺序死锁（Lock-Ordering DeadLock）。

# 1.死锁 #

> 死锁定义：当一个线程永远地持有一个锁，并且其他线程都尝试获取这个锁的时候，那么它们将永远被阻塞。在线程A持有L锁并且想获得M锁的同时，线程B持有M锁并且想获得L锁的时候，那么这两个先将永远等待下去，这就是最简单的死锁形式。（抱死锁）其中多个线程由于存在环路的锁依赖关系而永远地等待下去。

JVM在解决死锁的问题方面没有数据库服务那么强大。当一组线程发生死锁的时候，”游戏“到此结束————这些线程永远不能再使用了。根据线程完成的工作不同，可能造成应用程序完全停止，或者某个特定的子系统停止，或者是性能降低。恢复应用程序的唯一方式就是终止并且重启它，并希望不要再发生同样的事情。

## 1.1 锁顺序问题 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70]

public class LeftRightDeadLock { 
        private final Object left=new Object();
        private final Object right=new Object();
    
        private CountDownLatch countDownLatch=new CountDownLatch(1);
    
        public void leftRight(){ 
            try { 
                countDownLatch.await();
                synchronized (left){ 
                    Thread.sleep(1000);
                    synchronized (right){ 
                        System.out.println("-------leftRight---------");
                    }
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public void rightLeft(){ 
            try { 
                countDownLatch.await();
                synchronized (right){ 
                    Thread.sleep(1000);
                    synchronized (left){ 
                        System.out.println("-------Rightleft---------");
                    }
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public void countDown(){ 
            countDownLatch.countDown();
        }
    
        public static void main(String[] args) { 
            LeftRightDeadLock leftRightDeadLock=new LeftRightDeadLock();
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    leftRightDeadLock.leftRight();
                }
            }).start();
    
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    leftRightDeadLock.rightLeft();
                }
            }).start();
    
            leftRightDeadLock.countDown();
        }
    }

在LeftRightDeadLock中发生死锁的原因是：两个线程以不同的顺序来获得相同的锁。如果按照相同的顺序来请求锁，那么久不会出现循环的加锁依赖性，因此也就不会产生死锁了。

## 1.2 动态顺序死锁 ##

有时候我们并不清楚地知道是否在锁顺序上有足够的控制权来避免死锁。

线程不安全
    public class TransferMoney { 
        public void transferMoney(Account fromAccount, Account toAccount){ 
            synchronized (fromAccount){ 
                synchronized (toAccount){ 
                    //do transferMoney
                    System.out.println("--------transferMoney--------");
                }
            }
        }
    }

解决方案：我们可以通过调整锁顺序来避免死锁。比如下面：

public class ThreadSafeTransferMoney { 
        public static final Object tileLock=new Object();
        
        public void transferMoney(Account fromAccount, Account toAccount){ 
            int fromHash=System.identityHashCode(fromAccount);
            int toHash=System.identityHashCode(toAccount);
            
            if(fromHash > toHash){ 
                synchronized (fromAccount){ 
                    synchronized (toAccount){ 
                        //do transferMoney
                        System.out.println("--------transferMoney--------");
                    }
                }
            }else if(fromHash < toHash){ 
                synchronized (fromAccount){ 
                    synchronized (toAccount){ 
                        //do transferMoney
                        System.out.println("--------transferMoney--------");
                    }
                }
            }else { 
                synchronized (tileLock) { 
                    synchronized (fromAccount) { 
                        synchronized (toAccount) { 
                            //do transferMoney
                            System.out.println("--------transferMoney--------");
                        }
                    }
                }
            }
        }
    }

## 1.3 在协作对象之间发生死锁 ##

public class Point { 
        private int x;
        private int y;
    
        public Point(int x, int y) { 
            this.x = x;
            this.y = y;
        }
    
        public Point() { 
        }
    
        public int getX() { 
            return x;
        }
    
        public void setX(int x) { 
            this.x = x;
        }
    
        public int getY() { 
            return y;
        }
    
        public void setY(int y) { 
            this.y = y;
        }
    
        @Override
        public String toString() { 
            return "Point{" +
                    "x=" + x +
                    ", y=" + y +
                    '}';
        }
    }
    
    public class Taxi { 
        private final Dispatcher dispatcher;
    
        private Point location;
    
        private Point destination;
    
        public Taxi(Dispatcher dispatcher) { 
            this.dispatcher = dispatcher;
        }
    
        public synchronized Point getLocation(){ 
            return location;
        }
    
        private synchronized void setLocation(Point location){ 
            this.location=location;
            if(location.equals(destination)){ 
                dispatcher.notifyAvaiable(this);
            }
        }
    }
    
    public class Dispatcher { 
        private final Set<Taxi> taxis;
        private final Set<Taxi> availableTaxis;
    
        public Dispatcher() { 
            taxis=new HashSet<>();
            availableTaxis=new HashSet<>();
        }
    
        public synchronized void notifyAvaiable(Taxi taxi){ 
            availableTaxis.add(taxi);
        }
    
        public synchronized List<Point> getImage(){ 
            List<Point> list=new ArrayList<>();
            for (Taxi taxi : taxis) { 
                list.add(taxi.getLocation());
            }
            return list;
        }
    }

Taxi调用setLocation方法需要先获取Taxi的锁，然后再获得Dispatcher的锁。Dispatcher的getImage方法需要先获取Dispatcher的锁，然后再获取Taxi的锁，这与LeftRightDeadLock中的情况相同，两个线程按照不同的顺序获取锁，因此可能造成死锁。

> 如果在持有锁的情况下调用某个外部方法，那么将出现活跃性问题。在这个外部方法中可能会获取其他的锁（这可能造成死锁），或者阻塞时间过长，导致其他线程无法及时获得当前被持有的锁。为了解决这个协同对象之间的调用死锁问题，下面我们就将介绍开发调用方法。

## 1.4 开放调用 ##

在Taxi和Dispatcher中并不知道它们要陷入死锁，况且本来它们就不应该知道。方法调用相当于一种抽象屏障，因而你无须了解在被调用方法中所执行的操作。但也正是由于不知道在被调用方法中执行的操作，因此在持有锁的时候对调用某个外部方法难以进行分析，所以可能会导致出现死锁。

总结：如果我们能够在调用方法的时候不需要持有锁，那么这种调用就是一种开放调用。依赖于开放调用的类通常能表现出更好的行为，并且与那些在调用方法时需要持有锁的类相比，也更容易编写。

public class Taxi { 
        private final Dispatcher dispatcher;
    
        private Point location;
    
        private Point destination;
    
        public Taxi(Dispatcher dispatcher) { 
            this.dispatcher = dispatcher;
        }
    
        public synchronized Point getLocation(){ 
            return location;
        }
    
        private void setLocation(Point location){ 
            boolean readchedDestination=false;
    
            synchronized (this){ 
                this.location=location;
                if(location.equals(destination)){ 
                    readchedDestination=true;
                }
            }
    
            if (readchedDestination)
                dispatcher.notifyAvaiable(this);
        }
    }
    
    public class Dispatcher { 
        private final Set<Taxi> taxis;
        private final Set<Taxi> availableTaxis;
    
        public Dispatcher() { 
            taxis=new HashSet<>();
            availableTaxis=new HashSet<>();
        }
    
        public synchronized void notifyAvaiable(Taxi taxi){ 
            availableTaxis.add(taxi);
        }
    
        public List<Point> getImage(){ 
            List<Point> list=new ArrayList<>();
            Set<Taxi> copy=new HashSet<>();
    
            synchronized (this){ 
                copy.addAll(taxis);
            }
    
            for (Taxi taxi : copy) { 
                list.add(taxi.getLocation());
            }
            return list;
        }
    }

## 1.5 资源死锁 ##

> 定义：正当多个线程相互持有彼此正在等待的锁而又不释放自己已持有的锁时会发生死锁，当他们相同的资源集合等待时，也会发生死锁。

如果一个任务需要连接两个数据库，并且在请求这两个资源时不会始终遵循相同的顺序，那么线程A可能持有与数据库D1的连接，并等待与数据库D2的连接，而线程B则持有与D2的连接并等待与D1的连接（资源池越大，出现这种情况的可能性就越小，如果每个资源池都有N个连接，那么在发生死锁时不仅需要N个循环等待的线程，而且还需要大量不恰当的执行时序）

另一个资源的死锁形式就是线程饥饿死锁（Thread-Starvation Deadlock）。 比如个任务提交另一个任务，并等待被提交任务在单线程的Executor中执行完成。这种情况天，第一个任务将永远等待下去，并使得另一个任务以及在这个Executor中执行的所有其他任务都停止执行。

# 2.死锁的避免与诊断 #

如果必须获取多个锁，那么在设计时必须考虑锁的顺序：尽量减少潜在的加锁 交互数量，将获取锁时需要遵循的协议写入正式文档并始终遵循这些协议。

死锁避免地方法：

1.  首先，找出在什么地方将获取多个锁（使这个集合尽量小）
2.  然后对所有这些实例进行全局分析，从而确保它们在整个程序中获取锁的顺序都保持一致。
3.  尽可能地使用开放调用，这能极大地简化分析过程。

如果所有的调用都是开放调用，那么要发现获取多个锁的实例是非常简单的，可以通过代码审查或者借助自动化的源代码分析工具。

## 2.1 支持定时的锁（Timed Lock Attempts） ##

当使用内置锁时，只要没有获得锁，就会永远等待下去，而显式锁则可以执行一个超时时限（Timeout），在等待超过该事件后tryLock会返回一个失败信息。

即使在整个系统中没有始终使用定时锁，使用定时锁来获取多个锁也能有效地应对死锁问题。 如果在获取锁时超时，那么可以释放这个锁，然后后退并在一段时间后再次蚕食，从而消除了死锁发生的条件，使程序恢复过来。（这项技术只有在同时获取两个锁时才有效，如果在嵌套的方法调用中请求多个锁，那么即使你知道已经有了外层的锁，也无法释放它）

## 2.2 通过线程转储信息来分析死锁（Deadlock Analysis with Thread Dumps） ##

JVM通过线程转储（Thread Dump）来帮助识别死锁的发生。 线程转储包括各个运行中的线程的栈追踪信息，这类似于发生异常时的栈追踪信息。 线程转储还包含加锁信息，例如每个线程持有了哪些锁，在那些栈帧中获得这些锁，以及被阻塞的线程正在等待获取哪一个锁。

在生成线程转储之前，JVM将在等待关系图中通过搜索循环来找出死锁。如果发现了一个死锁，则获取相应的死锁信息，例如在死锁中涉及哪些锁和线程，以及这个锁的获取操作位于程序的哪些位置.

# 3. 活跃性危险 #

死锁时最常见的活跃性危险，在并发线程中还存在一些其他的活跃性危险，包括：饥饿，丢失信号和活锁等。

## 3.1 饥饿（Starvation） ##

**当线程由于无法访问它所需要的资源而不能继续执行时，就发生了“饥饿（Starvation）”。**

引发饥饿的最常见资源就是CPU时钟周期。如果在Java应用程序中对线程的优先级使用不当，或者在持有锁时执行一些无法结束的结构（例如无限循环，或无限制等待某个资源），那么也可能导致饥饿，因为其他需要这个锁的线程将无法得到它。

在Thread API定义的线程优先级只是作为线程调度的参考。在Thread API中定义了10个优先级，JVM根据需要将它们映射到操作系统的调度优先级，这种映射时与特定平台（不同的操作系统）相关的。在某些操作系统中，如果优先级的数量少于10个，那么有多个Java优先级会被映射到同一个优先级。

> 要避免使用线程优先级，因为这会增加平台依赖性，并可能导致活跃性问题。在大多数并发应用程序中，都可以使用默认的线程优先级。

## 3.2 糟糕的响应性（Poor Responsiveness） ##

CPU密集型的后台任务仍可能对响应性造成影响，因为它们会与事件线程共同竞争CPU的时钟周期。

解决方案：如果由其他线程完成的工作都是后台任务，那么应该降低它们的优先级，从而提高前台程序的响应性。

## 3.3 活锁 ##

> 活锁定义：活锁（Livelock）时另一种形式的活跃性问题，尽管不会阻塞线程，但也不能继续执行，因为线程不断重复执行相同的操作，而且总会失败。

1.  活锁通常发生在处理事务消息的应用程序中：如果不能成功地处理某个消息，那么消息处理机制将回滚整个事务，并将它重新放到队列的开头。如果消息处理其在处理某种特定类型的消息时存在错误并导致它失败，那么每当这个消息从队列中取出并传递到存在错误的处理器时，都会发生事务回滚。由于这条消息又被放回到队列开头，因此处理器将被反复调用，并返回先沟通的结果（有时候也被称为毒药消息，Poison Message）。**这种形式的活锁通常时由过度的错误恢复代码造成的，因为它错误将不可修复的错误作为可修复的错误。**
2.  当多个相互协作的线程都对彼此进行响应从而修改各自的状态，并使得任何一个线程都无法继续执行时，就发生了活锁。 这就像两个过于礼貌的人在半路上面对面相遇了：他们彼此都让出对方的路，然后又在另一条路上相遇了，因此他们就这样反复地避让下去。

解决方案：要解决这种活锁问题，需要在重试机制中引入随机性（randomness）。为了避免这种情况发生，需要让它们分别等待一段随机的时间

**在并发应用程序中，通过等待随机长度的时间和回退可以有效地避免活锁的发生。**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMTI1MTgz_size_16_color_FFFFFF_t_70]: /images/20220328/458a8d80867841be90ec21deefdadfb8.png