【Java并发编程系列9】锁

死锁死锁是指一组互相竞争资源的线程因互相等待导致“永久”阻塞的现象。锁顺序死锁我们先看一个死锁的示例我们先定义个BankAccount对象来存储基本信息代码如下public class BankAccount { private int id; private double balance; private String password; public int getId() { return id; } public void setId(int id) { this.id id; } public double getBalance() { return balance; } public void setBalance(double balance) { this.balance balance; } }接下来我们使用细粒度锁来尝试完成转账操作public class BankTransferDemo { public void transfer(BankAccount sourceAccount, BankAccount targetAccount, double amount) { synchronized(sourceAccount) { synchronized(targetAccount) { if (sourceAccount.getBalance() amount) { System.out.println(Start transfer.); sourceAccount.setBalance(sourceAccount.getBalance() - amount); targetAccount.setBalance(targetAccount.getBalance() amount); } } } } }如果进行下述调用就会产生死锁transfer(myAccount, yourAccount, 10); transfer(yourAccount, myAccount, 10);如果执行顺序不当那么A可能获取myAccount的锁并等待yourAccount的锁然而B此时持有yourAccount的锁并正在等待myAccount的锁。通过顺序来避免死锁由于我们无法控制参数的顺序如果要解决这个问题必须定义锁的顺序并在整个应用程序中按照这个顺序来获取锁。我们可以通过Object.hashCode返回的值来定义锁的顺序public class BankTransferDemo { private static final Object tieLock new Object(); public void transfer(BankAccount sourceAccount, BankAccount targetAccount, double amount) { int sourceHash System.identityHashCode(sourceAccount); int targetHash System.identityHashCode(targetAccount); if (sourceHash targetHash) { synchronized(sourceAccount) { synchronized(targetAccount) { if (sourceAccount.getBalance() amount) { sourceAccount.setBalance(sourceAccount.getBalance() - amount); targetAccount.setBalance(targetAccount.getBalance() amount); } } } } else if (sourceHash targetHash) { synchronized(targetAccount) { synchronized(sourceAccount) { if (sourceAccount.getBalance() amount) { sourceAccount.setBalance(sourceAccount.getBalance() - amount); targetAccount.setBalance(targetAccount.getBalance() amount); } } } } else { synchronized (tieLock) { synchronized(targetAccount) { synchronized(sourceAccount) { if (sourceAccount.getBalance() amount) { sourceAccount.setBalance(sourceAccount.getBalance() - amount); targetAccount.setBalance(targetAccount.getBalance() amount); } } } } } } }无论你入参怎么变化通过hash值的大小我们永远是先锁住hash值小的数据再锁hash值大的数据这样就保证的锁的顺序。但是在极少数情况下两个对象的Hash值相同如果顺序错了仍可能导致死锁所以在获取两个锁之前使用“加时赛Tie-Breaking”锁保证每次只有一个线程以未知的顺序获取到该锁。但是如果程序经常出现Hash冲突的情况这里会成为并发的瓶颈因为final变量是内存可见会让所有的线程都阻塞到该锁上不过这种概率会很低。在协作对象之间发生死锁这里我就只简单说明一下就是有两个对象A和BA.action_A1()会调用B中的方法action_B1()同时B.action_B2()会调用A中的方法action_A2()由于这四个方法action_A1()、action_A2()、action_B1()、action_B2()都通过synchronized加锁我们知道都通过synchronized在方法上加的是对象锁所以可能存在A调用B的方法时B也正在调用A的方法导致互相等待出现死锁的情况。具体的示例大家可以参考《Java并发编程实战》书籍第174页的内容。ReentrantLock使用方法在协调对象的访问时可以使用的机制只有synchronized和volatileJava 5.0增加了一种新的机制ReentrantLock。ReentrantLock并不是一种替代内置锁的方法而是当内置锁机制不适用时作为一种可选的高级功能。下面看一个简单的示例Lock lock new ReentrantLock(); //... lock.lock(); try { // ... } finally { lock.unlock(); }除了上述不可替换synchronized的原因就是需要手动通过lock.unlock()释放该锁如果忘记释放那将是个非常严重的问题。通过tryLock避免顺序死锁还是沿用上面的死锁示例我们通过tryLock()进行简单改造public boolean transfer(BankAccount sourceAccount, BankAccount targetAccount, double amount, long timeout, TimeUnit unit) { long stopTime System.nanoTime() unit.toNanos(timeout); while (true) { if (sourceAccount.lock.tryLock()) { try { if (targetAccount.lock.tryLock()) { try { if (sourceAccount.getBalance() amount) { sourceAccount.setBalance(sourceAccount.getBalance() - amount); targetAccount.setBalance(targetAccount.getBalance() amount); } } finally { targetAccount.lock.unlock(); } } } finally { sourceAccount.lock.unlock(); } } if (System.nanoTime() stopTime) { return false; } // Sleep一会... } }我们先尝试获取sourceAccount的锁如果获取成功再尝试获取targetAccount的锁如果获取失败我们就释放sourceAccount的锁避免长期占用sourceAccount锁而导致的死锁问题。带有时间限制的加锁我们也可以对tryLock()指定超时时间如果等待的时间超时不会一直等待直接执行后续的逻辑long stopTime System.nanoTime() unit.toNanos(timeout); while (true) { long nanosToLock unit.toNanos(timeout); if (sourceAccount.lock.tryLock(nanosToLock, TimeUnit.NANOSECONDS)) { try { // 省略... } finally { sourceAccount.lock.unlock(); } } if (System.nanoTime() stopTime) { return false; } // Sleep一会... }synchronized vs ReentrantLockReentrantLock在加锁和内存上提供的语义与内置锁相同此外它还提供了一些其他的功能包括定时的锁等待、可中断的锁等待、公平性以及实现非块结构的加锁。ReentrantLock的性能上似乎优于内置锁其中在Java 6.0中略有胜出而在Java 5.0中则远远胜出那是否我们都用ReentrantLock直接废弃掉synchronized么与显示锁相比内置锁仍然具有很大的优势。内置锁为许多开发人员所熟悉并且简洁紧凑。ReentrantLock的危险性比同步机制要高如果忘记在finally块中调用unlock那么虽然代码表面上看起来能正常运行但实际上已经埋下了一颗定时炸弹并很有可能伤及其它代码。仅当内置锁不能满足需求时才可以考虑使用ReentrantLock。使用原则ReentrantLock可以作为一种高级工具当需要一些高级功能比如可定时的、可轮训与可中断的锁获取操作公平队列以及非块结构的锁。否则还是优先使用synchronized。然后有一点需要重点强调一下synchronized和ReentrantLock都是可重入锁可重入的概念请参考文章《【Java并发编程系列3】synchronized》。读写锁读写锁的使用和Go中的读写锁用法一致先看读写锁接口定义public interface ReadWriteLock { /** * 返回读锁 */ Lock readLock(); /** * 返回写锁 */ Lock writeLock(); }ReadWriteLock管理一组锁一个是只读的锁一个是写锁。Java并发库中ReetrantReadWriteLock实现了ReadWriteLock接口并添加了可重入的特性。下面看一下使用姿势public class ReadWriteMapK,V { private final MapK,V map; private final ReadWriteLock lock new ReentrantReadWriteLock(); private final Lock r lock.readLock(); private final Lock w lock.writeLock(); public ReadWriteMap(MapK,V map) { this.map map; } public V put(K key, V value) { w.lock(); try { return map.put(key,value); } finally { w.unlock(); } } public V get(Object key) { r.lock(); try { return map.get(key); } finally { r.unlock(); } } }这样可以多个线程去读取数据但是只有一个线程可以去写数据然后读和写不能同时进行。其它自旋锁这个仅作为扩展知识觉得有些意思就写进来那么什么是自旋锁呢自旋锁的定义当一个线程尝试去获取某一把锁的时候如果这个锁此时已经被别人获取(占用)那么此线程就无法获取到这把锁该线程将会等待间隔一段时间后会再次尝试获取。这种采用循环加锁 - 等待的机制被称为自旋锁(spinlock)。自旋锁的原理自旋锁的原理比较简单如果持有锁的线程能在短时间内释放锁资源那么那些等待竞争锁的线程就不需要做内核态和用户态之间的切换进入阻塞状态它们只需要等一等(自旋)等到持有锁的线程释放锁之后即可获取这样就避免了用户进程和内核切换的消耗。因为自旋锁避免了操作系统进程调度和线程切换所以自旋锁通常适用在时间比较短的情况下。由于这个原因操作系统的内核经常使用自旋锁。但是如果长时间上锁的话自旋锁会非常耗费性能它阻止了其他线程的运行和调度。线程持有锁的时间越长则持有该锁的线程将被 OS(Operating System) 调度程序中断的风险越大。如果发生中断情况那么其他线程将保持旋转状态(反复尝试获取锁)而持有该锁的线程并不打算释放锁这样导致的是结果是无限期推迟直到持有锁的线程可以完成并释放它为止。解决上面这种情况一个很好的方式是给自旋锁设定一个自旋时间等时间一到立即释放自旋锁。自旋锁的优缺点自旋锁尽可能的减少线程的阻塞这对于锁的竞争不激烈且占用锁时间非常短的代码块来说性能能大幅度的提升因为自旋的消耗会小于线程阻塞挂起再唤醒的操作的消耗这些操作会导致线程发生两次上下文切换但是如果锁的竞争激烈或者持有锁的线程需要长时间占用锁执行同步块这时候就不适合使用自旋锁了因为自旋锁在获取锁前一直都是占用 cpu 做无用功占着 XX 不 XX同时有大量线程在竞争一个锁会导致获取锁的时间很长线程自旋的消耗大于线程阻塞挂起操作的消耗其它需要 cpu 的线程又不能获取到 cpu造成 cpu 的浪费。所以这种情况下我们要关闭自旋锁。自旋锁的实现public class SpinLockTest { private AtomicBoolean available new AtomicBoolean(false); public void lock(){ // 循环检测尝试获取锁 while (!tryLock()){ // doSomething... } } public boolean tryLock(){ // 尝试获取锁成功返回true失败返回false return available.compareAndSet(false,true); } public void unLock(){ if(!available.compareAndSet(true,false)){ throw new RuntimeException(释放锁失败); } } }这种简单的自旋锁有一个问题无法保证多线程竞争的公平性。对于上面的 SpinlockTest当多个线程想要获取锁时谁最先将available设为false谁就能最先获得锁这可能会造成某些线程一直都未获取到锁造成线程饥饿。就像我们下课后蜂拥的跑向食堂下班后蜂拥地挤向地铁通常我们会采取排队的方式解决这样的问题类似地我们把这种锁叫排队自旋锁(QueuedSpinlock)。计算机科学家们使用了各种方式来实现排队自旋锁如TicketLockMCSLockCLHLock。锁的特性Java 中的锁有很多可以按照不同的功能、种类进行分类下面是我对 Java 中一些常用锁的分类包括一些基本的概述从线程是否需要对资源加锁可以分为“悲观锁”和“乐观锁”从资源已被锁定线程是否阻塞可以分为“自旋锁”从多个线程并发访问资源也就是Synchronized可以分为无锁、偏向锁、轻量级锁和重量级锁从锁的公平性进行区分可以分为“公平锁”和“非公平锁”从根据锁是否重复获取可以分为“可重入锁”和“不可重入锁”从那个多个线程能否获取同一把锁分为“共享锁”和“排他锁”