《java并发编程实战》第十四章构建自定义的同步工具

朱雀 2022-03-20 06:54 238阅读 0赞

### 《java并发编程实战笔记》 ###

*  第十四章 构建自定义的同步工具
 *   *  状态依赖性的管理
     *   *  依赖条件不满足时，可将依赖条件的失败传递给调用者
         *  依赖条件不满足时，可通过轮询和休眠实现简单阻塞
         *  条件队列能及时响应依赖状态的改变且不浪费CPU
     *  如何使用条件队列
     *   *  条件等待的标准形式
         *  条件等待的通知 notify 与notifyAll
     *  显式的Condition对象提供更加细粒度的条件队列
     *  使用ReentrantLock实现信号量的练习
     *  AbstractQueuedSynchronizer(了解)
     *  小结

# 第十四章 构建自定义的同步工具 #

## 状态依赖性的管理 ##

**像FutureTask、Semaphore和BlockingQueue类中的操作有基于状态为前提的类称为状态依赖性类**。例如，不能从一个空的队列中删除元素、或者获取一个尚未结束的任务的计算结果。(依赖状态的个人理解：某个操作依赖于/等着什么状态条件改变，状态依赖性的管理需要考虑：在依赖条件不满足时该干什么？如何监测依赖状态的改变不浪费CPU资源，响应性高？)  
  例如在生产者—消费者的设计经常会像ArrayBlockingQueue的有界缓存，在有界缓存的put和take操作中都有状态依赖条件：不能从空缓存中获取元素，也不能将元素放入已满缓存中。**当依赖条件不满足时，这些依赖状态的操作put、take可以抛出一个异常或者返回一个错误，也可保持阻塞直到对象进入正确的状态。**

### 依赖条件不满足时，可将依赖条件的失败传递给调用者 ###

下面通过BaseBoundedBuffer有界缓存对依赖条件的失败不同处理介绍有界缓存的几种实现。

@ ThreadSafe
    public abstract class BaseBoundedBuffer<E> { 
           @GuardeBy( "this" ) private final E[] buf;
           @GuardeBy( "this" ) private int tail;
           @GuardeBy( "this" ) private int head;
           @GuardeBy( "this" ) private int count;
           protected BaseBoundedBuffer( int capacity) { 
                 this .buf = (E[]) new Object[capacity];
          }
           protected synchronized final void doPut(E E) { 
                buf[tail] = E;
                 if (++tail == buf.length) { 
                      tail = 0;
                }
                ++count;
          }
           protected synchronized final E doTake() { 
                E E = buf[head];
                buf[head] = null ;
                 if (++head == buf.length) { 
                      head = 0;
                }
                --count;
                 return E;
          }
           public synchronized final boolean isFull() { 
                 return count == buf.length;
          }
           public synchronized final boolean isEmpty() { 
                 return count == 0;
          }
    }

GrumpyBoundedBuffer将前提条件的失败传递给调用者，虽然这种方法实现起来很简单，但是调用者必须做好捕获异常的准备，并且当如果在多个地方都要调用put和take方法时，并且前提条件还失败，这样就要不断的重试。

@ ThreadSafe
    public class GrumpyBoundedBuffer<V> extends BaseBoundedBuffer<V> { 
           public GrumpyBoundedBuffer( int size){ 
                 super (size);
          }    
           public synchronized void put(V v){ 
                 if (isFull()){ 
                       throw new BufferFullException ();
                }
                doPut(v);
          }      
           public synchronized V take(){ 
                 if (isEmpty())
                       throw new BufferEmptyExeption ();
                 return doTake();
          }
    }

调用者要调用GrumpyBoundedBuffer的take和put方法，取出值进行操作，可以采用两种方式：1、在循环中不断重试，这种方法称为忙等待或者自旋等待。自旋导致CPU时钟周期浪费。2、当缓存状态不满足时，进行休眠。低响应性。

while (true ){ //自旋不断重试
                 try { 
                      V item = buffer.take();
                       // 对于item执行一些操作
                       break ;
                } catch (BufferEmptyException e) { 
                      Thread. sleep(SLEEP_GRANULARITY );
                }
          }

自旋有稍微好点的方法，在不满足依赖条件时，调用Thread.yield，相当于告诉调度器：现在需要让出一定的时间使另一个线程运行。当然即使使用了thread.yield()来让渡，但也只是减缓对CPU，上下文切换的消耗。

while(!message)
    { 
        Thread.yield()；
    }

### 依赖条件不满足时，可通过轮询和休眠实现简单阻塞 ###

SleepyBoundedBuffer就是通过“轮询与休眠”重试机制实现put、take方法，从而使调用者无需每次调用都实现重试的逻辑。如果依赖条件不满足，那么当前执行的线程首先会释放锁并且休眠一段时间，从而让其他线程能够访问缓存。当线程醒来时，它将重新请求锁并重新尝试操作，**因而线程能反复在休眠以及测试状态条件过程中切换**，直到可以执行位置。  
  从调用者看来，这种方法能很好的运行。但是如何选择合适的休眠时间间隔，就要在响应性和CPU使用率之间权衡，休眠时间间隔越小，响应性越高，但消耗的CPU资源越高。

@ ThreadSafe
    public class SleepyBoundedBuffer<V> extends BaseBoundedBuffer<V> { 
           public SleepyBoundedBuffer( int size) { 
                 super (size);
          }
           public void put(V v) throws InterruptedException{ 
                 while (true ){ 
                       synchronized (this ){ 
                             if (!isFull()){ 
                                  doPut(v);
                                   return ;
                            }
                      }
                      Thread.sleep(SLEEP_GRANULARITY);
                }
          }    
           public V take() throws InterruptedException{ 
                 while (true ){ 
                       synchronized (this ){ 
                             if (!isEmpty()){ 
                                   return doTake();
                            }
                      }
                      Thread.sleep(SLEEP_GRANULARITY);
                }
          }
    }

### 条件队列能及时响应依赖状态的改变且不浪费CPU ###

条件队列：装入的数据项是等待先验条件成立而被挂起的线程。不同于传统的队列存放数据。每个java对象可以作为一个锁，每个对象同样可以作为一个条件队，并且Object中的wait、notify和notifyAll方法就构成内部条件队列的api。对象的内置锁和内部条件队列是相互关联的，要调用某个对象中条件队列的wait、notify和notifyAll任何一个方法，必须先持有该对象上的锁。  
  Object.wait会自动释放锁，同时当前对象会请求操作系统挂起当前线程，此时对象的对象锁就可用了，允许其余等待线程进入。  
  “条件队列中的线程一定是执行不下去了才处于等待状态”,这个"执行不下去的条件"叫做"条件谓词。  
  需要注意的是，**wait()方法的返回并不一定意味着正在等待的条件谓词变成真了**。举个列子：假设现在有三个线程在等待同一个条件谓词变成真，然后另外一个线程调用了notifyAll()方法。此时，只能有一个线程离开条件队列，另外两个线程将仍然需要处于等待状态，**这就是在代码中使用while(conditioin is not true)\{this.wait();\}而不使用if(condition id not true)\{this.wait();\}的原因**。另外一种情况是：同一个条件队列与多个条件谓词互相关联。这个时候，当调用此条件队列的notifyAll()方法时，某些条件谓词根本就不会变成真。**综上，这就是为什么每当线程被从wait中唤醒时，都必须再次测试条件谓词。**  
  下面的BoundedBuffer中，当执行notifyAll()方法时，线程会从wait()地方唤醒(本来是不满足条件谓词，执行wait然后挂起了线程)。使用条件队列会明显比“休眠”有界缓存更加高效，响应性也更高。

@ ThreadSafe
    public class BoundedBuffer<V> extends BaseBoundedBuffer<V> {  
           // 条件谓词：not-full (!isFull())
           // 条件谓词：not-empty (!isEmpty()) 
           public BoundedBuffer( int size) { 
                 super (size);
          }   
           // 阻塞并直道：not-full
           public synchronized void put(V v) throws InterruptedException{ 
                 while (isFull()){ 
                      wait();
                }
                doPut(v);
                notifyAll();
          }     
           // 阻塞并直道：not-empty
           public synchronized V take() throws InterruptedException{ 
                 while (isEmpty()){ 
                      wait();
                }
                V v = doTake();
                notifyAll();
                 return v;
          }
    }

## 如何使用条件队列 ##

### 条件等待的标准形式 ###

锁、条件谓词、条件队列三者关系并不复杂，但是wait方法返回并不意味着线程正在等待的条件谓词变真了，一个条件队列与多个条件谓词相关是很常见的情况。因此，每次线程都从wait方法唤醒，都必须再次测试条件谓词，由于现场在条件谓词不为真的情况下也可以反复醒来，因此必须在一个循环中调用wait，并且每次迭代中都测试条件谓词。条件等待的标准形式如下：

void stateDependentMethod() throws InterruptedException{ 
    	synchronized(lock){ 
    		while(!conditionPredition)
    			lock.wait();
    		doSomething();
    	}
    }

### 条件等待的通知 notify 与notifyAll ###

信号丢失：指线程必须等待一个已经为真的条件，但在开始等待之前没有检测条件谓词。好比：启动了烤面包机去拿报纸，当烤面包机铃声响了，你没有听到还在等待烤面包机的铃声，因此可能会等待很长的时间。信号丢失也是一种活跃性故障。  
  条件等待的包括等待和通知。在BoundedBuffer中，在缓存变非空时，为了使take解除阻塞，必须确保每条使缓存变非空的代码路径都发出一个通知。在BoundedBuffer中，只有一条代码路径即put方法，因此在put成功一个元素到缓存后，将调用notifyAll。同样在成功take一个元素后也要调用notifyAll，向正在等待“不为满”条件的线程发出通知：缓存已经不满了。  
  通知是采用notifyAll和notify方法，无论调用哪个，都必须持有与条件队列对象相关的锁。调用notify时，JVM会从条件队列中等待的多个线程选择一个来唤醒，而调用notifyAll会唤醒这个条件队列上等待的所有线程。由于多个线程可以基于不同条件谓词在同一个条件队列上等待，那么如果使用notify而不是notifyAll将是一种危险的操作，容易导致信号丢失。  
  只有同时满足下面两个条件，才能用notify而不是notifyAll。但是大多数的类并不满足下面两个条件，因此普遍认可做法是notifyAll，虽然会比notify低效，但是可以确保正确。  
  1、只有一个条件谓词与条件队列相关，并且每个线程从wait返回后都执行相同的操作。  
  2、在条件变量上的每次通知，最多只能唤醒一个线程来执行。

## 显式的Condition对象提供更加细粒度的条件队列 ##

相较第一版的BoundedBuffer改进后，

public class BoundedBuffer<T> extends BaseBoundedBuffer<T> { 
    	public BoundedBuffer(int capacity){ 
    		super(capacity);
    	}
    	public synchronized void put(T t) throws InterruptedException{ 
    		while(!isFull())//执行该方法的对象在条件谓词 !isFull 上等待
    			this.wait();
    		this.doPut(t);
    		this.notifyAll();
    	}
    	public synchronized T get() throws InterruptedException{ 
    		while(!isEmpty())//执行该方法的对象在条件谓词 !isEmpty 上等待
    			this.wait();
    		T t=this.doGet();
    		this.notifyAll();
    		return t;
    	}
    }

然而在上面的BoundedBuffer例子中，同一个条件队列上存在两个条件谓词。这样，当调用notifyAll()方法的时候，唤醒的不仅是在!isFull()上等待的线程，还有在!isEmpty()上等待的线程，尽管唤醒在!isEmpty()上等待的线程是没有必要的，这就迫使我们想使用一种更加细颗度的条件队列。在Java中，除了提供内置锁和内置条件队列，还提供显式锁和显式条件队列。其中显式锁为Lock，显示条件队列为Condition对象。

一个Condition是和一个Lock关联起来的，就像一个内置条件队列和一个内置锁关联起来一样。**要创建一个Condition，可以在相关联的Lock上调用newCondition()方法**。每个内置锁只能有一个与之关联的内置条件队列，与之不同的是，**每个Lock上可以有多个与他关联的Condition,这就使得我们对Condition的控制更加细粒度化**。 对于上面的BoundedBuffer类，使用显式条件队列进行改进，如下：(signal比signalAll更高效，能极大的减少在每次缓存操作中发生的上下文切换和锁请求的次数。)

import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    public class ConditionBoundedBuffer<T>  { 
    	private Lock lock=new ReentrantLock();
    	private Condition notFull=lock.newCondition();
    	private Condition notEmpty=lock.newCondition();	
    	private int head,tail,count;
    	private Object[] buf;
    	
    	public ConditionBoundedBuffer(int capacity){ 
    		buf=new Object[capacity];
    		head=0;tail=0;count=0;
    	}	
    	public void add(T t) throws InterruptedException{ 
    		lock.lock();
    		try{ 
    			while(count==buf.length)
    				notFull.wait();
    			buf[tail]=t;
    			if(++tail==buf.length)
    				tail=0;
    			count++;
    			notEmpty.signal();
    		}finally{ 
    			lock.unlock();
    		}
    	}
    	public T get() throws InterruptedException{ 
    		lock.lock();
    		try{ 
    			while(count==0)
    				notEmpty.wait();
    			Object obj=buf[head];
    			buf[head]=null;
    			if(++head==buf.length)
    				head=0;
    			count--;
    			notFull.signal();
    			return (T)obj;
    		}finally{ 
    			lock.unlock();
    		}
    	}
    }

## 使用ReentrantLock实现信号量的练习 ##

在ReentrantLock和Semaphore两个接口中存在很多共同点，两个类都可以做一个“阀门”，每次都只允许一定数量线程通过，两者其实都是使用了一个共同的基类，即AbstractQueuedSynchronizer(AQS)，ReentrantLock和Semaphore都实现了Synchronizer。  
一个常见的练习，使用Lock来实现计数信号量。

@ThreadSafepublic class SemaphoreOnLock {     
    	private final Lock lock = new ReentrantLock();    // CONDITION PREDICATE: permitsAvailable (permits > 0)
        private final Condition permitsAvailable = lock.newCondition();    @GuardedBy("lock") private int permits;
        SemaphoreOnLock(int initialPermits) { 
            lock.lock();        try { 
                permits = initialPermits;
            } finally { 
                lock.unlock();
            }
        }    // BLOCKS-UNTIL: permitsAvailable
        public void acquire() throws InterruptedException { 
            lock.lock();        try {             while (permits <= 0)
                    permitsAvailable.await();
                --permits;
            } finally { 
                lock.unlock();
            }
        }    public void release() { 
            lock.lock();        try { 
                ++permits;
                permitsAvailable.signal();
            } finally { 
                lock.unlock();
            }
        }
    }

AQS (AbstractQueuedSynchronizer)是一个构建锁和同步器的框架，CountDownLatch、SynchronousQueue和FutureTask也是由AQS构建的。

## AbstractQueuedSynchronizer(了解) ##

大多数的开发者都不会直接用AQS，常见的标准同步器类集合能满足大多数的需求。java.util.concurrent中许多可阻塞的类，例如ReentrantLock、Semaphore、ReentrantReadWriteLock、CountDownLatch、SynchronousQueue和FutureTask都是基于AQS构建的，不需要过于深入了解实现的细节。

## 小结 ##

要实现一个包含依赖状态的类－－如果没有满足依赖状态的前提条件，那么这个类的方法必须阻塞，那么最好的方式是基于现有的库类来构建，例如CountDownLatch、Semaphore。例如第八章的ValueLatch所示。然而有时，现有的库类不能提供足够的功能，这种情况下，可以使用内置的条件队列、显式的Condition对象或者AbstractQueuedSynchronizer来构建自己的同步器，显式的Condition和Lock相比内置条件队列提供了扩展的功能集，包括每个锁对应多个等待线程集、可中断或者不可中断的条件等待，公平或者非公平队列操作。