BiBi - 并发编程 -10- 队列同步器

From：Java并发编程的艺术

• 目录
  BiBi - 并发编程 -0- 开篇
  BiBi - 并发编程 -1- 挑战
  BiBi - 并发编程 -2- volatile
  BiBi - 并发编程 -3- 锁
  BiBi - 并发编程 -4- 原子操作
  BiBi - 并发编程 -5- Java内存模型
  BiBi - 并发编程 -6- final关键字
  BiBi - 并发编程 -7- DCL
  BiBi - 并发编程 -8- 线程
  BiBi - 并发编程 -9- ReentrantLock
  BiBi - 并发编程 -10- 队列同步器
  BiBi - 并发编程 -11- 并发容器
  BiBi - 并发编程 -12- Fork/Join框架
  BiBi - 并发编程 -13- 并发工具类
  BiBi - 并发编程 -14- 线程池
  BiBi - 并发编程 -15- Executor框架

队列同步器简介

队列同步器【AbstractQueuedSynchronizer】是构建锁和其他同步组件的基础框架，它使用一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。通常会被定义为一个内部类来继承AQS。

基于AQS实现的同步器有：ReentrantLock、Semaphore、ReentrantReadWriteLock、CountDownLatch、FutureTask。

队列同步器实例

同步器提供的三个方法：
getState()
setState(int newState)
compareAndSetState(int expect, int update)

同步器是基于模板设计模式实现的，所以自定义同步器，重写相应的方法即可。在实现方法过程中，若对同步状态进行更改，要使用上述三个方法。

自定义同步组件：

package com.ljg.concurrent;

import android.support.annotation.NonNull;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

//自定义一个独占锁
public class Mutex implements Lock {

  //静态内部类，自定义同步器
  private static class Sync extends AbstractQueuedSynchronizer {
    //是否处于占用状态
    @Override
    protected boolean isHeldExclusively() {
      return getState() == 1;
    }

    //当前状态为0时获取锁，获取失败会被加入到同步队列中等待
    @Override
    protected boolean tryAcquire(int arg) {
      if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());//设置当前线程拥有独占访问权限
        return true;
      }
      return false;
    }

    //释放锁，将状态设置为0
    @Override
    protected boolean tryRelease(int arg) {
      if (getState() == 0) {
        throw new IllegalMonitorStateException();
      }
      setExclusiveOwnerThread(null);
      setState(0);
      return true;
    }

    //返回一个Condition，每个condition都包含一个condition队列
    Condition newCondition() {
      return new ConditionObject();
    }
  }

  private final Sync sync = new Sync();

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

  @Override
  public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
  }

  @Override
  public boolean tryLock() {
    return sync.tryAcquire(1);
  }

  @Override
  public boolean tryLock(long time, @NonNull TimeUnit unit) throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(time));
  }

  @Override
  public void unlock() {
    sync.release(1);
  }

  @NonNull
  @Override
  public Condition newCondition() {
    return sync.newCondition();
  }

  public boolean isLocked() {
    return sync.isHeldExclusively();
  }

  public boolean hasQueuedThreads() {
    return sync.hasQueuedThreads();
  }
}

可见，通过代理ASQ能够实现大部分同步需求。

队列同步器实现原理

内部维护一个FIFO双向队列，header节点分别指向队列中第一个节点和最后一个节点。获取同步状态失败的线程会被加入到队列的尾部并在队列中进行自旋，移除队列的条件是前驱节点为头节点且成功获取了同步状态。加入尾队列过程中，需要使用compareAndSetTail方法确保安全性；在设置首节点时则不需要，因为此时线程已得到了锁。

共享式获取与独立式获取最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。例如：写操作要求对资源的独占式访问，而读操作可以是共享式访问。共享式访问资源时，其它共享式的访问资源均被允许，而独占式访问被阻塞；独占式访问资源时，其它所有访问都被阻塞。

TwinsLock实例

要求：同一时刻只允许至多两个线程同时访问，超过两个线程的访问将被阻塞。

package com.ljg.concurrent;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Lock;

public class Mutex implements Lock {

  //静态内部类，自定义同步器
  private static class Sync extends AbstractQueuedSynchronizer {
    Sync(int count) {
      if (count <= 0) {
        throw new IllegalArgumentException("count must large than zero.");
      }
      setState(count);
    }

    public int tryAcquireShared(int reduceCount) {
      for (; ; ) {
        int current = getState();
        int newCount = current - reduceCount;
        if (newCount < 0 || compareAndSetState(current, newCount)) {
          return newCount;
        }
      }
    }

    public boolean tryReleaseShared(int returnCount) {
      for (; ; ) {
        int current = getState();
        int newCount = current + returnCount;
        if (compareAndSetState(current, newCount)) {
          return true;
        }
      }
    }
  }

  private final Sync sync = new Sync(2);
    @Override
    public void lock() {
        sync.acquireShared(1);
    }

    @Override
    public void unlock() {
        sync.releaseShared(1);
    }
  //省略接口的实现方法
}

在上例中TwinsLock实现了Lock接口，提供了面向使用者的接口，使用者调用lock()方法获取锁，使用unlock()释放锁，而且同一时刻只能有两个线程获取到锁。以共享式获取同步状态为例：同步器会先计算出获取后的同步状态，然后通过CAS确保状态的正确设置，当tryAcquireShared(int reduceCount)方法返回值大于等于0时，当前线程才能获取同步状态，对于上层的TwinsLock而言，则表示当前线程获取锁。

可见，同步器作为一个桥梁，连接了线程访问以及同步状态控制等底层技术与不同并发组件（Lock、CountDownLatch等）的接口语义。