[Java并发编程实战] 第14章构建自定义的同步工具-白红宇

[Java并发编程实战] 第14章构建自定义的同步工具

阅读量：2341 次

发布时间：2019-05-10

本文共 5015 字，大约阅读时间需要 16 分钟。

状态依赖性的管理

构成前提条件的状态变量必须由对象的锁来保护，从而使他们在测试前提条件的同时保持不变；如果前提条件尚未满足，就必须释放锁，以便其它线程可以修改对象的状态，否则，前提条件就永远无法变成真。在再次测试前提条件之前，必须重新获得锁

将前提条件的失败传递给调用者，调用者可以选择休眠等待、自旋等待或者调用Thread.yield

可以通过简单的“轮询与休眠”重试机制实现阻塞，同时将前提条件的管理操作封装起来

条件队列

使得一组线程（等待线程集合）能够通过某种方式来等待特定的条件变成真；条件队列中的元素是一个个正在等待相关条件的线程

每个Java对象可以作为一个锁，也可以作为一个条件队列，Object的wait/notify/notifyAll方法构成了内部条件队列的API。对象的内置锁与其内部条件队列是相互关联的，要调用对象X中条件队列的任何一个方法，必须持有对象X上的锁

Object.wait会自动释放锁，并请求操作系统挂起当前线程，从而使其他线程能够获得这个锁并修改对象的状态。当被挂起的线程醒来时，他将在返回之前重新获得锁

public synchronized void put(V v) throws InterruptedException{
           while(isFull()){
               System.out.println(new Date()+" buffer 满了， thread wait:"+Thread.currentThread().getName());            wait();        }        doPut(v);        System.out.println(new Date()+" "+Thread.currentThread().getName()+" 放入 ："+v+" ");        notifyAll();    }        public synchronized V take() throws InterruptedException {
           while(isEmpty()){
               System.out.println(new Date()+" buffer 为空， thread wait:"+Thread.currentThread().getName());            wait();        }                        notifyAll();                //每当在等待一个条件时，一定要确保在条件谓词变为真时，通过某种方式发出通知         V v = doTake();                System.out.println(new Date()+" "+Thread.currentThread().getName()+" 取出 :"+v);        return v;    }

条件谓词：是使某个操作成为状态依赖操作的前提条件；每当线程从wait中唤醒时，都必须再次测试条件谓词，因此需要在一个循环中调用wait

void stateDependentMethod() throws InterruptedException{
       synchronized(lock){
           while(!conditionPredicate())            lock.wait();        //现在对象处于合适的状态    }}

当使用条件等待时（Object.wait/Condition.await）
- 通常有一个条件谓词——包括一些对象状态的测试，线程在执行前必须首先通过这些测试
- 在调用wait之前测试条件谓词，并且从wait中返回时再次进行测试
- 在一个循环中调用wait
- 确保使用与条件队列相关的锁来保护构成条件谓词的各个状态变量
- 当调用wait/notify/notifyAll等方法时，一定要持有与条件队列相关的锁
- 在检查条件谓词之后以及开始执行相应的操作之前，不要释放锁

调用notify时，JVM会从这个条件队列上等待的多个线程中选择一个来唤醒（单一的通知很容易导致信号丢失），而调用notifyAll则会唤醒所有在这个条件队列上等待的线程

只有同时满足以下两个条件时，才能用单一的notify:
- 所有等待线程的类型都相同：只有一个条件谓词与条件队列相关，并且每个线程在从wait返回后将执行相同的操作
- 单进单出：在条件变量上的每次通知，最多只能唤醒一个线程来执行

用“入口协议”和“出口协议”描述wait和notify方法的正确使用：
- 对于每个依赖状态的操作，以及每个修改其他操作依赖状态的操作，都应该定义一个入口协议和出口协议
- 入口协议：就是该操作的条件谓词
- 出口协议：包括检查该操作修改的所有状态变量，并确认它们是否使某个其它的条件谓词变为真，如果是，则通知相关的条件队列

显式的Condition对象

Lock是一种广义的内置锁，Condition也是一种广义的内置条件队列

与内置条件队列不同的是，对于每个Lock，可以有任意数量的Condition对象；Condition比内置条件队列提供了更丰富的功能：在每个锁上可存在多个等待、条件等待可以是可中断的或不可中断的、基于时限的等待，以及公平的或非公平的队列操作

当使用显式的Lock和Condition时，必须满足锁、条件谓词、条件变量之间的三元关系：条件谓词中包含的变量必须由Lock保护，并且在检查条件谓词以及调用await和signal时，必须持有Lock对象

/** * 使用显式条件变量的有界缓存 */public class ConditionBoundedBuffer
   
     {
       protected final Lock lock = new ReentrantLock();    //条件谓词：notFull (count < items.length)    private final Condition notFull = lock.newCondition();    //条件谓词：notEmpty (count > 0)    private final Condition notEmpty = lock.newCondition();    private static final int BUFFER_SIZE = 100;    @GuardedBy("lock")    private final T[] items = (T[]) new Object[BUFFER_SIZE];    @GuardedBy("lock")    private int tail,head,count;        //阻塞并直到：notFull    public void put(T x) throws InterruptedException {
           lock.lock();        try{
               while(count == items.length)                notFull.await();            items[tail] = x;            if(++tail == items.length)                tail = 0;            ++count;            notEmpty.signal();        }finally{
               lock.unlock();        }    }        //阻塞并直到：notEmpty    public T take() throws InterruptedException{
           lock.lock();        try{
               while(count==0)                notEmpty.await();            T x = items[head];            items[head] = null;            if(++head == items.length)                head = 0;            --count;            notEmpty.signal();            return x;        }finally{
               lock.unlock();        }    }}

AbstractQueuedSynchronizer

LOCK的实现类其实都是构建在AbstractQueuedSynchronizer上，每个Lock实现类都持有自己内部类Sync的实例，而这个Sync就是继承AbstractQueuedSynchronizer(AQS)。

提供 volatile 变量 state;用于同步线程之间的共享状态。通过 CAS 和 volatile 保证其原子性和可见性。

/**  * 同步状态  */  private volatile int state;    /**  *cas  */  protected final boolean compareAndSetState(int expect, int update) {
         // See below for intrinsics setup to support this      return unsafe.compareAndSwapInt(this, stateOffset, expect, update);  }

同步器是实现锁的关键，利用同步器将锁的语义实现，然后在锁的实现中聚合同步器。可以这样理解：锁的API是面向使用者的，它定义了与锁交互的公共行为，而每个锁需要完成特定的操作也是透过这些行为来完成的（比如：可以允许两个线程进行加锁，排除两个以上的线程），但是实现是依托给同步器来完成；同步器面向的是线程访问和资源控制，它定义了线程对资源是否能够获取以及线程的排队等操作。锁和同步器很好的隔离了二者所需要关注的领域，严格意义上讲，同步器可以适用于除了锁以外的其他同步设施上（包括锁）。

java.util.concurrent同步器类中的AQS

java.util.concurrent中的很多可阻塞类都是基于AQS构建的

ReentrantLock:
- 只支持独占方式的获取操作，实现了tryAcquire/tryRelease/isHeldExclusively
- 将同步状态用于保存锁获取操作的次数，并且维护一个owner变量保存当前所有者线程的标识符

Semaphore:
- 将AQS的同步状态用于保存当前可用许可的数量

CountDownLatch:
- 同步状态中保存的是当前的计数值

FutureTask:
- AQS同步状态用来保存任务的状态：正在运行、已完成、已取消
- 还维护一些额外的状态变量，保存计算结果或抛出异常
- 还维护一个引用，指向正在执行计算任务的线程，因而任务取消，线程中断

ReentrantReadWriteLock:
- 内部单个AQS子类同时管理读取加锁和写入加锁
- 使用一个16位的状态表示写入锁的计数，使用另一个16位的状态表示读取锁的计数
- 读取锁的操作使用共享的获取和释放方法，写入锁使用独占的获取释放方法
- AQS在内部维护一个等待线程队列，记录了某个线程请求的是独占访问还是共享访问
- 当锁可用时，如果位于队列头部的线程执行写入操作，那么线程会得到这个锁；如果位于队列头部的线程执行读取访问，那么队列中第一个写入线程之前所有的线程都将获得这个锁