java并发编程实战-第14章-构建自定义的同步器

java并发编程实战-第14章-构建自定义的同步器

14.构建自定义的同步器

类库中包含很多状态依赖的类 FutureTask 、Semaphore 和 BlockingQueue等

14.1 状态依赖的管理

  在并发程序中，基于状态的条件可能会由于其他线程的操作而改变

  

  通过轮询和sleep可以勉强解决状态依赖的问题，但高效的做法是使用条件等待机制

  

14.1.1 将前提条件的失败传给调用者 。

        这导致调用者还得自己处理前提条件失败的情况

14.1.2 通过轮询和sleep可以勉强解决状态依赖的问题

       解决状态依赖的问题

 

14.1.3  条件队列

       wait  notify/notofyAll

14.2  使用条件队列

        条件队列使构建搞笑及可响应的状态类变得更容易，但同时也很容易被不正确使用。使用一定规则

 

可以保证正确使用，但平台和编译器并没有强制要求

        遵循这些规则。

        

       This is one of the reasons to build on top of classes like LinkedBlockingQueue, 

 

CountDown-Latch, Semaphore, and FutureTask when you can; if you can get away with it, it is a 

 

lot easier

14.2.1 条件谓词

     1、 Document the condition predicate(s) associated with a condition queue and the 

 

operations that wait on them.

      

     2、Every call to wait is implicitly associated with a specific condition predicate. When 

 

calling wait regarding a particular condition predicate, the caller must already hold the lock 

 

associated with the condition queue, and that lock must also guard the state variables from 

 

which the condition predicate is composed.

        

        重要的三元关系： 加锁 、条件谓词、和wait（） 

      

      

        比如 BoundedBuffer.java

      

   // BLOCKS-UNTIL: not-full

   public synchronized void put(V v) throws InterruptedException { //加锁 

       while (isFull())//条件谓词

           wait(); //wait  

       doPut(v);

       notifyAll();

   }

 

   // BLOCKS-UNTIL: not-empty

   public synchronized V take() throws InterruptedException {

       while (isEmpty())

           wait();

       V v = doTake();

       notifyAll();

       return v;

   }

      

14.2.2 过早唤醒

        当一个线程由于待用notifyAll而醒来时，并不意味着该线程等待的条件谓词已经变成真了。所以

 

呢，wait 要放在一个循环里，当唤醒时，再次判断一下条件

        

        

        When using condition waits (Object.wait or Condition.await):

 

Always have a condition predicatesome test of 

 

object state that must hold before proceeding;

 

Always test the condition predicate before 

 

calling wait, and again after returning from wait;

 

Always call wait in a loop;

 

Ensure that the state variables making up the 

 

condition predicate are guarded by the lock associated with the condition queue;

 

Hold the lock associated with the the 

 

condition queue when calling wait, notify, or notifyAll; and

 

Do not release the lock after checking the 

 

condition predicate but before acting on it.

       如上总结步骤：锁->循环等待条件if（!condition）{wait}->执行相关操作->notify/notifyAll

       

14.2.3 丢失的信号 

 

比如 ：fails to check the condition predicate before waiting

 

；没在条用wait之前检查条件

 

14.2.4 通知

       Single notify can be used instead of notifyAll only when both of the following 

 

conditions hold:

Uniform waiters（相同类型的等待者）. Only one condition 

 

predicate is associated with the condition queue, and each thread executes the same logic upon 

 

returning from wait; and

One-in, one-out（单进单出）. A notification on the condition 

 

variable enables at most one thread to proceed.

 

优化：条件通知： 很难实现，原则：首先使程序真确执行，然后使其更快

 

Listing 14.8. Using Conditional Notification in BoundedBuffer.put.

public synchronized void put(V v) throws InterruptedException 

 

{

   while (isFull())

       wait();

   boolean wasEmpty = isEmpty();

   doPut(v);

   if (wasEmpty)

       notifyAll();

}

 

    因为，在take线程则只有在wasEmpty=true的时候阻塞，所以可以优化为当wasEmpty=true时，put的操

 

作后才notify。

    

14.2.5   示例阀门类

 

  一个允许打开后能马上就关闭的阀门类，线程通过的条件（isOpen 或者 已经被打开过，但是现在关闭了

 

）

 

Listing 14.9. Recloseable Gate Using Wait and Notifyall.

@ThreadSafe

public class ThreadGate {

   // CONDITION-PREDICATE: opened-since(n) (isOpen || generation>n)

   @GuardedBy("this") private boolean isOpen;

   @GuardedBy("this") private int generation;

 

   public synchronized void close() {

       isOpen = false;

   }

 

   public synchronized void open() {

       ++generation;

       isOpen = true;

       notifyAll();

   }

 

   // BLOCKS-UNTIL: opened-since(generation on entry)

   public synchronized void await() throws InterruptedException {

       int arrivalGeneration = generation;

       while (!isOpen && arrivalGeneration == generation)

           wait();

   }

}

 

===流程分析

  close()   isOpen = false; arrivalGeneration=0,generation=0

  t1.await  isOpen = false; arrivalGeneration=0,generation=0

  t2.await  isOpen = false; arrivalGeneration=0,generation=0

  open()    isOpen = true;  arrivalGeneration=0,generation=1

  t1 t2 通过 （isOpen=true）

  close()   isOpen = false; arrivalGeneration=1,generation=1

  t3.await  isOpen = false; arrivalGenerati=1,generation=1

  t4.await  isOpen = false; arrivalGenerati=1,generation=1

  open()    isOpen = true; arrivalGenerati=1,generation=2

  close()   isOpen = false; arrivalGeneration=1,generation=2

  t2 t3 通过 （isOpen=false,但是arrivalGeneration<generation，说明t3、t4在await的时候，已经打

 

开了，所以可以通过）

  

 

14.2.6 子类的安全问题

 

  要么围绕继承来设计和文档化 ，要么完全禁止子类化，例如将类声明为final，或者将条件队列、锁和状

 

态变量隐藏起来

     

  安全问题例子：比如子类添加了一个阻塞连续弹出2个元素的方法，则父类的push方法必须在子类重写为

 

notifyAll

14.2.7 封装条件队列

    使用私有锁，不再支持客户端加锁。通常应该是这样子的，但是这和线程安全类的常见设计模式并不一

 

致。即缓存对象即是锁又是缓存队列

    

14.2.8  入口协议和出口协议

   入口协议 ：条件谓词condition predicate

   出口协议 ：检查操作是否改变状态，使其他条件谓词为真，是则通知

   

   AbstractQueuedSynchronizer, 是有出口协议的，但不是自己notify，通过显示的api，比如getState（

 

）。明确的API调用，使得其在状态转化是更不容易被忘记通知。

   

   

14.3 显示的Condition对象

     参考 ：ConditionBoundedBuffer

     // CONDITION PREDICATE: notFull (count < items.length)

       private final Condition notFull = lock.newCondition();

       notFull.await();是怎么实现线程阻塞与锁释放呢？

       

    实现类AbstractQueuedSynchronizer$ConditonObject->await()->  LockSupport.park(this)-

 

>setBlocker(t, blocker);  unsafe.park(false, 0L);   setBlocker(t, null);

       

14.4 Synchronizer剖析

 

    CountDownLatch, ReentrantReadWriteLock, SynchronousQueue 和  FutureTask.都使用了共同的基类

 

：AbstractQueuedSynchronizer

    

14.5 AbstractQueuedSynchronizer

    

     重点：理解其工作原理

     

     基本操作 acquire and release 

     

     acquire：Acquisition is the state-dependent operation and can always block

     

     

     release： 更新同步状态，如果新的允许阻塞的线程获取成功。则解除阻塞状态

       

     Listing 14.13. Canonical Forms for Acquisition and Release in AQS.

     

boolean acquire() throws InterruptedException {

   while (state does not permit acquire) {

       if (blocking acquisition requested) {

           enqueue current thread if not already queued

           block current thread

       }

       else

           return failure

   }

   possibly update synchronization state

   dequeue thread if it was queued

   return success

}

 

void release() {

   update synchronization state

   if (new state may permit a blocked thread to acquire)

       unblock one or more queued threads

}

 

 

 

   基类 AQS 封装了acquire() 和 release()，

   

   基类中acquire() 和 release()调用的子类中用带try前缀的方法来判断某个操作是否会成功。（如果成

 

功则进行入队、阻塞等操作）  在同步器的子类中，根据具体语义，使用getState和setState以及

 

compareAndSetState来检查和更新状态，

   并通过返回的状态值来判断基类的acquire、release操作是否成功

   

   参考例子 ：OneShotLatch 

   

14.6  AQS in Java.util.concurrent Synchronizer Classes

 

14.6.1 ReentrantLock   （重点分析）  

       只支持独占方式获取锁，主要是实现tryAcquire, tryRelease, and isHeldExclusively，以 

 

tryAcquire（）分析

14.6.2 Semaphore and CountDownLatch

        共享，待分析     

14.6.3 FutureTask

       待具体分析

14.6.4 ReentrantReadWriteLock

       待具体分析

 

小结：

  

   要实现一个状态依赖的类，最好的方式是基于现有的类，比如Semaphore, BlockingQueue, or 

 

CountDownLatch。如果跟多要求，则通过AQS自定义自己的同步器类