JAVA多线程之AQS分析(3)

public class ReentrantReadWriteLockExample {
  private static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

  private static  double data=0;
  private static ExecutorService executor = Executors.newFixedThreadPool(20);

  static class readClass implements Callable<Integer> {
      private int i;
      public readClass(int i) {this.i = i;}
      
      @Override
      public Integer call() throws Exception {
        rwl.readLock().lock();
        System.out.println(i + " -->读数据："+data);
        rwl.readLock().unlock();
        
        return 0;
      }
  }

  static class writeClass implements Callable<Integer> {

    private double i;
    public writeClass(double i) {this.i = i;}

    @Override
    public Integer call() throws Exception {
      rwl.writeLock().lock();
      data=i;

      System.out.println("写数据： "+data);
      try {
        System.out.printf("sleep: ...");
         Thread.sleep(3000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      
      System.out.printf("wake up ... ");
      rwl.writeLock().unlock();
      
      return 1;
    }
  }
  public static void main(String[] args) throws InterruptedException {
    List<Callable<Integer>> callList = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        readClass readClass = new readClass(i);
        callList.add(readClass);
    }
    callList.add(new writeClass(10));
    for (int i = 10; i < 20; i++) {
      readClass readClass = new readClass(i);
      callList.add(readClass);
    }
    List<Future<Integer>> futures = executor.invokeAll(callList);
    executor.shutdown();
  }
}

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable {

    private final ReentrantReadWriteLock.ReadLock readerLock;
    private final ReentrantReadWriteLock.WriteLock writerLock;

    final Sync sync;

    public ReentrantReadWriteLock(){
        this(false);
    }

    public ReentrantReadWriteLock(boolean fair){
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
    //......
}

abstract static class Sync extends AbstractQueuedSynchronizer {
  
  static final int SHARED_SHIFT   = 16;
  static final int SHARED_UNIT    = (1 << SHARED_SHIFT);      // 000000000 00000001 00000000 00000000
  static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;  // 000000000 00000000 11111111 11111111
  
  //表示计算写锁的具体值时，该值为 15个1,用 getState & EXCLUSIVE_MASK算出写锁的线程数，大于1表示重入。
  static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; 

  /** 计算 readLock 的获取次数(包含 reentrant 的次数) */
  static int sharedCount(int c)       { return c >>> SHARED_SHIFT; } // 将字节向右移动 16位, 只剩下 原来的 高 16 位
  
  /** 计算 writeLock 的获取的次数(包括 reentrant的次数) */
  static int exclusiveCount(int c)    { return c & EXCLUSIVE_MASK; } // 与 EXCLUSIVE_MASK 与一下
// ...略

public static class ReadLock implements Lock, java.io.Serializable {

  private static final long serialVersionUID = -5992448646407690164L;
  private final Sync sync;
  protected ReadLock(ReentrantReadWriteLock lock) {
      sync = lock.sync;
  }
  public void lock() {sync.acquireShared(1);}
  // ... other
  public void unlock() {sync.releaseShared(1);}
}

// sync 的获取共享锁的方法
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);
    // 这里表示 读锁是否应该阻塞 这里的方法是根据 公平和非公平的锁来实现的
    // 公平锁 是直接调用 hasQueuedPredecessors() 
    // 非公平锁 直接调用的是  apparentlyFirstQueuedIsExclusive(); 方法
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        // 读锁的数量 + 1
        compareAndSetState(c, c + SHARED_UNIT)) {
        // 加锁之前 没有读锁 本线程是第一个
        if (r == 0) {
            // 记录第一个读锁的线程
            firstReader = current;
            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;
    }
    // 上面的条件路径都失败，那么会继续调用fullTryAcquireShared方法。  
    return fullTryAcquireShared(current);
}

static final class HoldCounter {
      int count = 0;
      final long tid = getThreadId(Thread.currentThread());
  }
  static final class ThreadLocalHoldCounter
      extends ThreadLocal<HoldCounter> {
      public HoldCounter initialValue() {
          return new HoldCounter();
      }
  }
  private transient ThreadLocalHoldCounter readHolds;
  private transient HoldCounter cachedHoldCounter;

final int fullTryAcquireShared(Thread current) {
  HoldCounter rh = null;
  for (;;) {
    int c = getState();
    if (exclusiveCount(c) != 0) {
      if (getExclusiveOwnerThread() != current)
          return -1;
      /* 如果当前持有写锁的是本线程，那么应该让继续去获取读锁，否则一旦返回，改线程睡眠，则会导致死锁 */
    } else if (readerShouldBlock()) {
      // 这里的 firstReader 的作用就出来了，这里firstReader的缓存可以有效的帮助
      // 快速的判断 读线程是否需要阻塞
      if (firstReader == current) {
        // assert firstReaderHoldCount > 0;
      } else {
        /* 
          这里的代码是本线程应该排队的情况，如果最后一个获取读锁的线程不是本线程
          意味着，这个读线程应该排队。
          如果该线程的数量读锁的数量为0 证明没有以前并没有获取过，那么本线程应该睡眠
          否则一旦发现 本线程确实有获取过读锁，那么不能睡眠（不能返回-1），重入锁要走的自选逻辑。
        */
          if (rh == null) {
            // 上一个读锁的线程
            rh = cachedHoldCounter;
            // 如果没有人获取过读锁，或者上一个获取读锁的不是本线程
            if (rh == null || rh.tid != getThreadId(current)) {
                // 拿到当前线程的计数器
                rh = readHolds.get();
                // 如果计数器 ==0 证明 该读锁已经没有获取过任何锁了，直接可以进入阻塞逻辑
                if (rh.count == 0)
                    readHolds.remove();
              }
          } 
          // 如果计数器 ==0 证明 该读锁已经没有获取过任何锁了，直接可以进入阻塞逻辑
          if (rh.count == 0)
              return -1;
        }
      }
      // 一下是正常的加锁逻辑，共享锁首锁缓存，数量+1 计数器+1 等等
      if (sharedCount(c) == MAX_COUNT)
        throw new Error("Maximum lock count exceeded");
      if (compareAndSetState(c, c + SHARED_UNIT)) {
        if (sharedCount(c) == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            firstReaderHoldCount++;
        } else {
            if (rh == null)
                rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
            rh.count++;
            cachedHoldCounter = rh; // cache for release
        }
        return 1;
    }
  }
}

if (!readerShouldBlock() &&r < MAX_COUNT &&
    compareAndSetState(c, c + SHARED_UNIT)) {
    //...
  }

// 这段代码的含义是，如果当前线程拥有写锁，但是又要去申请写锁，是允许的当写锁被持有时，
// 如果持有该锁的线程不是当前线程，就返回 “获取锁失败”，反之就会继续获取读锁。称之为锁降级。
if (getExclusiveOwnerThread() != current)
      return -1;

private void doAcquireShared(int arg) {
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
      boolean interrupted = false;
      for (;;) {
          final Node p = node.predecessor();
          if (p == head) {
              int r = tryAcquireShared(arg);
              if (r >= 0) {
                  setHeadAndPropagate(node, r);
                  p.next = null; // help GC
                  if (interrupted)
                      selfInterrupt();
                  failed = false;
                  return;
              }
          }
          // 阻塞
          if (shouldParkAfterFailedAcquire(p, node) &&
              parkAndCheckInterrupt())
              interrupted = true;
      }
  } finally {
      if (failed)
          cancelAcquire(node);
  }
}

// ReentrantReadWriteLock 代码
public void unlock() {
    sync.releaseShared(1);
}

// Sync 
public final boolean releaseShared(int arg) {
  if (tryReleaseShared(arg)) {
      doReleaseShared();
      return true;
  }
  return false;
}

// AQS 代码
protected final boolean tryReleaseShared(int unused) {
    Thread current = Thread.currentThread();
    // 首个获取读锁的线程
    if (firstReader == current) {
        // assert firstReaderHoldCount > 0;
        // 读锁数量 -1 如果没有清空
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    } else {
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            readHolds.remove();
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        --rh.count;
    }
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        if (compareAndSetState(c, nextc))
            // Releasing the read lock has no effect on readers,
            // but it may allow waiting writers to proceed if
            // both read and write locks are now free.
            return nextc == 0;
    }
}

private void doReleaseShared() {
  
  for (;;) {
      Node h = head;
    
      if (h != null && h != tail) {
          int ws = h.waitStatus;
          if (ws == Node.SIGNAL) {
            // 如果头结点 是等待信号状态，那么意味着head 节点是阻塞的
              if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                  continue;            // loop to recheck cases
              unparkSuccessor(h);     // 唤醒线程
          }
          else if (ws == 0 &&
                  !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
              continue;                // loop on failed CAS
      }
      if (h == head)                   // loop if head changed
          break;
  }
}

public void lock() {
  sync.acquire(1);
}

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
}

static final class FairSync extends Sync {
    private static final long serialVersionUID = -2274990926593161451L;
    final boolean writerShouldBlock() {
        return hasQueuedPredecessors();
    }
    final boolean readerShouldBlock() {
        return hasQueuedPredecessors();
    }
}

static final class NonfairSync extends Sync {
  private static final long serialVersionUID = -8159625535654395037L;
  final boolean writerShouldBlock() {
      return false; // writers can always barge
  }
  final boolean readerShouldBlock() {
      /* As a heuristic to avoid indefinite writer starvation,
      * block if the thread that momentarily appears to be head
      * of queue, if one exists, is a waiting writer.  This is
      * only a probabilistic effect since a new reader will not
      * block if there is a waiting writer behind other enabled
      * readers that have not yet drained from the queue.
      */
      return apparentlyFirstQueuedIsExclusive();
  }
}

final boolean apparentlyFirstQueuedIsExclusive() {
  Node h, s;
  return (h = head) != null &&
      (s = h.next)  != null &&
      !s.isShared()         &&
      s.thread != null;
}