黑马程序员技术交流社区

标题: 【上海校区】Java并发包中的锁原理剖析 [打印本页]

作者: 不二晨 时间: 2019-2-22 16:44
标题: 【上海校区】Java并发包中的锁原理剖析
LockSupport工具类

LockSupport是JUC并发包下的一个工具类，它的底层是由Unsafe类实现的，它是基于Unsafe类的park()和unpark()方法包装并扩展功能实现的一个工具类，它主要用于对线程的挂起和唤醒操作，它是线程安全的，底层由CAS算法实现。通过它的名字我们也可以知道，它是锁的支持类，后面的锁的实现都是基于它实现的。
park()方法和wait()方法
先谈谈wait()和Unsafe类的park()方法，LockSupport的park()方法是对Unsafe类的park()方法的包装，所以我们需要先理解底层的实际挂起线程的park()方法。
Object类的wait()源码：

public final void wait() throws InterruptedException {
      wait(0);
}
public final native void wait(long timeout) throws InterruptedException;

Unsafe类的park()源码：

public native void park(boolean isAbsolute, long time);

**park()方法的许可证：**上述两种挂起线程的方法底层都是native方法，我们知道wait()方法可以让一个线程挂起并释放锁资源，前提是必须在synchronize关键字内调用，在别的地方调用会抛出 java.lang.IllegalMonitorStateException异常。而park()方法只是挂起线程并不涉及到锁，park()方法会关联一个许可证，这个许可证的获取方式是调用LockSupport的unpark()方法，并且把希望唤醒的线程对象传入，接下来有两种情况：一种是线程直接调用park()方法，此时它因为没有许可证而被挂起，然后调用unpark()方法并把这个线程对象作为参数，此时线程会被唤醒；另一种情况是先调用unpark()方法，传入该线程对象，再调用这个线程对象的park()方法，它在被阻塞之后会直接返回。许可证被赋予之后只能使用一次。
**许可证的抽象理解：**许可证的机制可以理解为一道关卡，所有线程到了这里都会停下来，它们会一直等待许可证的发放，此时有许可证的线程可以通过，许可证使用过后就会被销毁，分发许可证的方法是unpark()。于是，这里总共涉及到了两样事物，一个是关卡，也就是无参的park()方法；另一个是许可证的分发，也就是unpark()方法；
**许可证的意义：**许可证的存在使得park()方法和unpark()方法真正意义上成为并发包的基础，它可以构成锁，park()方法使得没有许可证的线程无法进入相应代码块，它会使得无关线程挂起，防止它们争夺资源，导致死锁。它和synchronize的设计效果是相同的，区别是synchronize是获取和释放对象锁，而park()与unpark()是获取许可证实现“通行”。park()与unpark()的发展前景更好，可以基于它设计线程的队列，例如后面要讲到的AQS队列，在这两个方法或者AQS队列的基础上实现各种不同功能的锁。
LockSupport的park()方法：
然后再来看看LockSupport的park()方法，它底层也调用了Unsafe的park()方法，源码如下：

LockSupport.park();
public static void park() {
   UNSAFE.park(false, 0L);
}

我知道unsafe的park方法需要两个参数的传入。true表示绝对时间，一般在使用的时候需要获取当前时间加上希望的偏移时间（ms）来作为第二项的参数；false表示相对时间，表示方法调用多少时间之后返回（ns）。在LockSupport中对park做了封装，默认参数为0，表示一直被挂起。
LockSupport的unpark()方法：
unpark方法调用的也就是Unsafe的unpark方法，需要有线程传入才有效。

LockSupport.unpark(current);
public static void unpark(Thread thread) {
   if (thread != null)
      UNSAFE.unpark(thread);
  }

**其他返回情况：**被挂起的线程也可能被一些别的情况唤醒，包括小概率的非正常唤醒，以及被中断唤醒。这里介绍中断唤醒，当一个线程被挂起的时候，被别的线程调用了它的interrupt()方法，它就会被唤醒，为了区别线程的唤醒原因，一般在唤醒后判断中断标志，如果不希望线程被中断唤醒，可以设计while()循环，唤醒后根据中断标志决定是退出循环还是重新挂起。如果只希望被中断唤醒，那么while的循环条件可以是while(!Thread.currentThread().isInterrupted)只有被唤醒后标志为true才会离开循环，否则继续挂起。总之，需要考虑到被中断唤醒这种情况。
LockSupport.parkNanos(nanos)方法：
挂起指定时间的方法，如果线程持有许可证，那么挂起后直接返回；如果线程没有许可证，挂起nanos时间后自动返回。
LockSupport.park(blocker)方法：
测试Domo：如下两段测试分别是无参park方法和传入this的park方法。

public class Test {
   public static void main(String[] args) {
   Test test=new Test();
   test.test();
   }
   public void test(){
   LockSupport.park();  //LockSupport.park(this);
   }
}
后者输出多了一段wait时间，即调用了park方法后的内部信息。为什么呢？我们来分析源码：
如下代码中在调用Unsafe的方法中设置了blocker，在被唤醒后会清除这个bolcker对象，初步推测这是个线程的成员变量，所以可以获取内部信息并且在使用完后要释放内存。最后调用的方法中传入了当前线程和blocker实例（传入的Test对象），
parkBlockerOffset在LockSupport类中有一个静态变量是用来保存传入的变量引用的。通过它可以保存更多堆栈信息，可以得知线程被阻塞后发生的事情。

public static void park(Object blocker) {
      Thread t = Thread.currentThread();
      setBlocker(t, blocker);
      UNSAFE.park(false, 0L);
      setBlocker(t, null);
}
private static void setBlocker(Thread t, Object arg) {
      // Even though volatile, hotspot doesn't need a write barrier here.
      UNSAFE.putObject(t, parkBlockerOffset, arg);
}

LockSupport.parkNanos(blocker, nanos)方法：
它和上者的区别的是它是一个有时间的阻塞方法，同时传入bolcker记录堆栈信息。
LockSupport.parkUntil(blocker, deadline);方法：
这个方法是指指定截止时间，到某个时间后自动唤醒，同样它也传入了一个blocker对象。底层Unsafe方法的park参数是true，表示绝对时间。

public static void parkUntil(Object blocker, long deadline) {
      Thread t = Thread.currentThread();
      setBlocker(t, blocker);
      UNSAFE.park(true, deadline);
      setBlocker(t, null);
}

AQS（抽象同步队列）

它是一个抽象类，它的作用是通过FIFO（先进先出）队列的形式对线程对资源的访问顺序进行管理，并维护一些参数记录信息。通过它可以实现各种类型的锁，因为它的一些方法没有具体实现，需要交给子类去完成。一般来说JUC包中的锁都是继承自它，然后实现功能的。
AQS的参数：

队列元素类型：Node，其中的thread成员变量用来保存线程的引用。
节点内部的SHARED（共享）和EXCLUSIVE（独有）分别表示节点被阻塞放入队列的原因。
waitStatus记录线程的等待状态，可以为CANCELLED（取消了）、SIGNAL（线程需要被唤醒）、CONDITION（线程在条件队列里等待）。
prev记录当前节点的前驱结点，next记录当前节点的后继节点。
AQS队列中还维持了一个单一的状态信息state，例如可重入锁的重入次数就是由它来记录的。如果它被实现为读写锁，那么它的高16位记录读锁的获取次数，低16位才用来记录重入次数。
锁的独占方式：
独占方式是指state表示重入次数，一个线程获取了锁之后，别的线程再试图去访问都会被挂起等待，获取的锁的线程会把锁的持有者成员变量设置为自己，同时把state设置为1，之后别的线程访问该资源，通过持有者线程引用比对后发现不是它，那么就会被挂起，之前的锁使用完资源后，会释放锁，这里是释放指的是清零state变量，同时把持有者线程置为null，最后再唤醒一个在AQS队列中等待的线程。
锁的共享方式：
共享方式可以理解为公共资源，资源可以被多个线程获取，但是资源有限。可以理解为一个停车场，它有三个停车位，来了五辆车，前三辆获取了停车位，另外两辆车只能被“挂起”等待。在锁的实现中，有Semaphore（信号量）。共享方式的特点就是规定了资源的获取上限，由state变量来保存信息，state变量可以理解为“资源”。
条件变量：
试想一下，如果线程去申请获取资源，都会被放入AQS阻塞队列，最终就都会获取到资源，只是说有一个先后顺序，那么对于线程的约束就很少了，要想控制他们获取资源的过程，需要加上一些“条件”。比如把某些不希望它们获取资源的线程对象阻塞住，不让它们进入AQS阻塞队列，但同时也希望保存它们的引用，因为在满足一些条件的情况下可以再让它们参与资源的“争夺”。这时可以引入条件变量，它就类似于wait和notify争夺的对象锁一样，当某个线程满足某种条件的时候，就把它阻塞并放到某个队列，这个队列不是AQS，而是条件变量队列，在需要的时候再唤醒它。那么也有很多条件约束的的情况发生，那么就可以有多个条件变量，之前说过条件变量就类似于对象锁一样，那么这个条件变量就是一个新的引用，它也有一个属于自己的队列，在合适的实机，唤醒它们，使得这些在队列中的线程可以进入AQS争夺资源。这里的条件变量是Condition类的实例，它的方法是await()和signal()，当然还有signalAll()的方法，等同与之前的wait()和notify()，notifyAll()方法。下面的图示是条件变量队列和AQS队列的关系：

可以发现，加上条件变量后，对于线程的管理有了极大的灵活性可控性。
基于AQS实现不可重入的独占锁并用它实现生产消费者模型

public class NoReentrantLock implements Lock,java.io.Serializable{
/**
   * 内部帮助类
   * @author mayifan
   *
   */
public static class Sync extends AbstractQueuedSynchronizer{
/**
   * 是否锁已经被持有
   */
      protected boolean isHeldExclusively() {
         return getState()==1;
      }
      /**
      * 尝试获取锁
      */
      protected boolean tryAcquire(int arg) {
         assert arg==1;//判断数据合法性，如果是1，继续运行，如果是其他的，则抛出异常
         if(compareAndSetState(0, 1)){
            setExclusiveOwnerThread(Thread.currentThread());//设置持有者为当前线程
            return true;
         }
         return false;
      }
      /**
      * 尝试释放锁
      */
      protected boolean tryRelease(int arg) {
         assert arg==1;
         if(getState()==0||Thread.currentThread()!=getExclusiveOwnerThread()){
            throw new IllegalMonitorStateException();
         }
         setExclusiveOwnerThread(null);
         setState(0);
         return true;
      }
      /**
      * 提供条件变量接口
      */
      Condition newCondition(){
        return new ConditionObject();
      }
}
   private final Sync sync=new Sync();//实例化一个同步器
/**
   * 加锁
   */
public void lock() {
sync.acquire(1);
}
/**
   * 加锁（会对中断响应）
   */
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
/**
   * 尝试加锁
   */
public boolean tryLock() {
return sync.tryAcquire(1);
}
/**
   * 尝试加锁（失败则挂起time时间，）
   */
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
      return sync.tryAcquireNanos(1, unit.toNanos(time));
}
/**
   * 解锁
   */
public void unlock() {
sync.release(1);
}
/**
   * 获取条件变量
   */
public Condition newCondition() {
return sync.newCondition();
}
      /**
      * 判断是否被锁
      * @return
      */
public boolean isLocked(){
return sync.isHeldExclusively();
}
}

**分析该锁：**该锁的设计思路是先在内部写一个内部类，一个同步器，同步器重写了AQS的一些方法，以实现不可重入锁的功能，其中state只能在0和1之间转化，如果一个线程已经获取了锁，再次调用lock()方法，返回false，设置失败。因为它是一个锁，实现了锁Lock的接口，即便sync已经可以实现所有功能，但是还是需要把它适配为一个锁，对外呈现的也是锁的基本方法。其中的lockInterruptibly()方法表示会对中断做出响应，我们来看源码：
进入这个方法后会判断线程的中断标志，如果为true，会抛出异常。如果为false，即没有被中断过，则继续下面的代码，先尝试获取资源，如果失败，就放入队列（此时中断状态是false的）。接下来该线程会不断尝试，先是判断它是不是在队列首位同时成功获取资源，获取成功就会返回，一旦获取失败就会在下面的parkAndCheckInterrupt()方法中挂起，如果它被中断，在判断中断标志后返回一个true，接着抛出异常，如果不是中断唤醒它，它会继续循环尝试。抛出的异常就是对中断的相应。

public final void acquireInterruptibly(int arg)
         throws InterruptedException {
      if (Thread.interrupted())
         throw new InterruptedException();
      if (!tryAcquire(arg))
         doAcquireInterruptibly(arg);
}
private void doAcquireInterruptibly(int arg)
      throws InterruptedException {
      final Node node = addWaiter(Node.EXCLUSIVE);
      boolean failed = true;
      try {
         for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                  setHead(node);
                  p.next = null; // help GC
                  failed = false;
                  return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                  parkAndCheckInterrupt())
                  throw new InterruptedException();
         }
      } finally {
         if (failed)
            cancelAcquire(node);
      }
}
private final boolean parkAndCheckInterrupt() {
      LockSupport.park(this);
      return Thread.interrupted();
}

用上面的锁实现一个生产消费者模型：

public class Test {
   final static NoReentrantLock lock=new NoReentrantLock();
   final static Condition notFull=lock.newCondition();
   final static Condition notEmpty=lock.newCondition();
   final static Queue<String> queue=new LinkedBlockingQueue<String>();
   final static int queueSize=10;
   public static void main(String[] args) {
      //生产者线程
   Thread producer=new Thread(new Runnable() {
public void run() {
while(true){
lock.lock();//获取独占锁
try{
//如果队列满了，则等待
while(queue.size()==queueSize){
notEmpty.await();//存入notEmpty的队列
}
//添加元素到队列
queue.add("ele");
Thread.sleep(200);
System.out.println("添加一个元素到队列，队列当前长度为："+queue.size());
//唤醒消费线程，消费者在notFull中
notFull.signalAll();
}catch(Exception e){
e.printStackTrace();
}finally{
lock.unlock();
}
}
}
});
   //消费者线程
   Thread consumer=new Thread(new Runnable() {
public void run() {
while(true){
   lock.lock();
   try{
   if(queue.size()==0){
   notFull.await();
   }
   queue.poll();
   Thread.sleep(200);
   System.out.println("从队列移除一个元素，队列当前长度为："+queue.size());
   notEmpty.signalAll();
   }catch(Exception e){
   e.printStackTrace();
   }finally{
   lock.unlock();
   }
}
}
});
//启动线程
   producer.start();
   consumer.start();
   }
}

输出：

**模型分析：**上面这个模型用到了之前设计的独占锁，还用到了两个条件变量，这两个条件变量分别存放被阻塞的生产者和消费者线程。基本的思路是当队列满的时候生产者会阻塞自己，把自己加入到notEmpty条件变量的队列，当它被唤醒的时候，它会被转移到AQS队列中准备获取锁，拿到锁就可以工作了，它被唤醒可以说明消费者已经移除了部分元素，于是它继续往生产队列中添加元素，然后通知所有消费者工作，消费者会从队列中唤醒，进入AQS队列准备获取锁，循环往复。上面的代码中出现了生产者和消费者竞争锁的现象，出现了连续被某一方拿到锁的情况，值得肯定的是，数据没有异常，整个过程是线程安全的。

独占锁ReentrentLock（可重入锁）原理

实现过了和ReentrentLock类似的NoReentrentLock，理解ReentrentLock就很容易了。它和NoReentrentLock的区别是此锁是可重入的，重入之后state加一，释放一次state减一。如果当前线程没有持有该锁而调用了释放锁的方法，就会抛出异常。二者的相同之处在于，它们都是独占的，锁被占用时，其他线程不能得到该锁。关于这个锁的应用，比如可以用它来实现一个线程安全的List，在List的每一部分代码前后加锁就可以了，至于为什么连读取的代码也要加锁？是为了避免读取时数据被修改而导致错误。

读写锁ReentrentReadWriteLock（可重入锁）原理

上述独占锁用于读少写多的情况，而这个读写锁更多地应用于读多写少的情况。读写锁依然可以利用AQS的state参数来实现，利用它的高16位记录读锁的获取次数，低16位记录写锁的可重入次数，这样就可以对读锁和写锁分别判断了。写锁是独占锁，一旦有线程修改state低16位从0到1，并设置其为获取写锁的线程，那么其他线程获取读锁和写锁都会失败，直到它释放写锁，如果写锁没有被占用，线程对读锁的获取一般都会成功，只要总数不超过读锁获取次数的上限就可以了。
几个参数：
firstReader记录第一个获取读锁的线程，firstReaderHoldCount记录第一个获取到读锁的线程的获取读锁的可重入次数，cachedHoldCounter记录最后一个获取读锁线程的可重入次数。另外每个线程内部都有一个readHolder变量，它是线程私有的，它存放除了第一个线程外其他线程获取读锁的可重入次数。另外state的高十六位记录读锁被获取的总次数，这个数字表示的当前的一个值，会随着锁的获取与释放而变化。
读写锁中的同步器：
读写锁中和其他锁一样，都有实现同步器，同步器Sync继承了AbstractQueuedSynchronizer。然后有两个同步器子类继承了Sync，一个是非公平的同步器，一个是公平的同步器。
非公平同步器：它对于竞争写锁的线程的判断永远返回false，这个方法如果返回true表示它应该被挂起，返回false表示不应该被强制挂起，因为这个方法一直返回false，所以这个线程对锁的竞争与其在队列中的位置无关，它有可能会比先于它进入队列的线程先获得锁，这就是非公平的。非公平体现在没有维护先来先得的规则。
公平同步器：区别是它调用了hasQueuedPredecessors();方法，它会判断它有无前驱结点，如果有，返回true，它会挂起自己，主动退出对锁的竞争。

/**
   * Nonfair version of Sync
   */
static final class NonfairSync extends Sync {
      private static final long serialVersionUID = -8159625535654395037L;
      final boolean writerShouldBlock() {
         return false; // writers can always barge
      }
      final boolean readerShouldBlock() {
         return apparentlyFirstQueuedIsExclusive();
      }
      final boolean apparentlyFirstQueuedIsExclusive() {
      Node h, s;
         return (h = head) != null &&
            (s = h.next)  != null &&
            !s.isShared()       &&
            s.thread != null;
      }
}
/**
   * Fair version of Sync
   */
static final class FairSync extends Sync {
      private static final long serialVersionUID = -2274990926593161451L;
      final boolean writerShouldBlock() {
         return hasQueuedPredecessors();
      }
      final boolean readerShouldBlock() {
         return hasQueuedPredecessors();
      }
}
public final boolean hasQueuedPredecessors() {
      Node t = tail; // Read fields in reverse initialization order
      Node h = head;
      Node s;
      return h != t &&
         ((s = h.next) == null || s.thread != Thread.currentThread());
}

ReentrentReadWriteLock中的读锁

public ReentrantReadWriteLock(boolean fair) {
      sync = fair ? new FairSync() : new NonfairSync();
      readerLock = new ReadLock(this);
      writerLock = new WriteLock(this);
}
1
2
3
4
5
在new读写锁的时候会同时实例化同步器，读锁和写锁，把同步器传入读锁和写锁的构造方法中，它们调用的方法都是在同步器中对AQS重写后的方法。写锁的原理和独占锁相同，我们来看看读锁的源码：
它的构造方法传入了读写锁的引用，把读写锁使用的同步器是引用给它。

      protected ReadLock(ReentrantReadWriteLock lock) {
         sync = lock.sync;
      }

lock()方法:（过程分析见代码中的注释）

   public void lock() {
         sync.acquireShared(1);
   }
   public final void acquireShared(int arg) {
         if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
   }
      protected final int tryAcquireShared(int unused) {
         //获取当前状态值
         Thread current = Thread.currentThread();
         int c = getState();
         //判断写锁是否被占用，如果被占用则返回
         if (exclusiveCount(c) != 0 &&
            getExclusiveOwnerThread() != current)
            return -1;
         //获取读锁计数值
         int r = sharedCount(c);
         //尝试获取锁，多线程下只有一个线程会成功
         if (!readerShouldBlock() &&
            r < MAX_COUNT &&
            compareAndSetState(c, c + SHARED_UNIT)) {
            if (r == 0) {
                  //把当前线程作为第一个获取读锁的线程
                  firstReader = current;
                  //记录可重入次数为1
                  firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                  //第二次进入就累加了
                  firstReaderHoldCount++;
            } else {
                  HoldCounter rh = cachedHoldCounter;
                  if (rh == null || rh.tid != getThreadId(current))
                     cachedHoldCounter = rh = readHolds.get();
                  else if (rh.count == 0)
                     readHolds.set(rh);
                  rh.count++;
            }
            return 1;
         }
         //获取失败，自旋重试，会有机会获得锁，受同时竞争锁的线程数制约
         return fullTryAcquireShared(current);
      }

void lockInterruptibly()方法：这个方法中线程如果被中断返回会抛出异常。
boolean tryLock()方法：尝试获取锁，获取失败返回false，不会被阻塞。如果该线程已经获得了读锁，则简单增加state的高16位并返回true。
释放锁的尝试释放锁的部分代码如下,这段代码的预计结果是使得state的高16位减一，实现释放锁，如果多个线程在做这件事，可能会设置失败，那么这时，需要不断自旋重试，直到成功，这里是一个死循环。

      protected final boolean tryReleaseShared(int unused) {
      ..............
         for (;;) {
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc))
                  return nextc == 0;
         }
      }

**关于实现List线程安全的demo优化：**之前提到用独占锁实现List的线程安全，在修改时就不能读取了。如果使用读写锁就会更加灵活一些，比如为读取数据的代码用lock.writeLock();加锁，为修改数据的代码用lock.readLock();加锁。这就实现了多线程读的效果，在读多写少的情况下效率很高。

JDK8中新增的StampedLock

这种锁实现了三种模式的读写控制，在获取锁的时候会返回一个Long类型的戳记stamp，在释放锁的时候需要传入那个返回的戳记，获取锁失败时返回的戳记是0。它的读写模式都是不可重入的。
三种读写模式的锁：

写锁writeLock：它是一个独占锁，它和读写锁中的写锁的区别是，它是不可重入的，并传递一个stamp参数，tryWriteLock方法是非阻塞的。
悲观读锁readLock：它是一个共享锁，在没有线程使用写锁的情况下，可以有多个线程获取读锁；一旦有线程获取写锁之后，它会使得申请获取该读锁的线程挂起，它悲观地认为获取写锁的线程会对数据进行修改，所有不允许其他线程获取读锁，这应用于读少写多的情况。它也是通过stamp来传递实现获取和释放锁的。
乐观读锁tryOptimisticRead：它没有使用CAS去修改参数，而是先通过位运算测试，如果当前没有线程获取写锁，则会返回一个stamp！=0的版本号，再检查一个这个版本号是否可用，如果可用，则复制一份数据到方法栈，进行操作。这种锁适合写少读多的情况，读到的数据是一个快照，可能在复制数据后数据被人修改了，那么数据就不是最新的了，但数据依然是具有一致性的。这种锁其实并不是真正意义上的锁，它没有用到CAS算法，但它读数据的效率很高。
转化： StampedLock可以实现三种模式的转化：①、写锁模式。②、读锁模式。③、乐观读锁模式，且写锁可用。
特点：三种锁都是不可重入锁，如果已经获得锁了再次尝试获得锁，有可能导致线程阻塞。多个线程获取锁的过程没有一定的规则，是随机的。该锁没有实现Lock或ReadWriteLock接口，而是在其内部维护了一个双向阻塞队列。它性能优于读写锁的最重要原因是它有乐观读锁这个不需要使用CAS算法的模式，实现数据的高效读取。
---------------------
【转载，仅作分享，侵删】
作者：mayifan_blog
原文：https://blog.csdn.net/mayifan_blog/article/details/87222847
版权声明：本文为博主原创文章，转载请附上博文链接！

作者: 不二晨 时间: 2019-2-26 15:33

奈斯，感谢分享

欢迎光临黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2