zoukankan      html  css  js  c++  java

  • condition

    condition 介绍及demo

    　Condition是在java 1.5中才出现的，它用来替代传统的Object的wait()、notify()实现线程间的协作，相比使用Object的wait()、notify()，使用Condition的await()、signal()这种方式实现线程间协作更加安全和高效。因此通常来说比较推荐使用Condition，阻塞队列实际上是使用了Condition来模拟线程间协作。

    • Condition是个接口，基本的方法就是await()和signal()方法；
    • Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition() 
    •  调用Condition的await()和signal()方法，都必须在lock保护之内，就是说必须在lock.lock()和lock.unlock之间才可以使用

    　　Conditon中的await()对应Object的wait()；

    　　Condition中的signal()对应Object的notify()；

    Condition中的signalAll()对应Object的notifyAll()。

    下面是demo：　　

    [java] view plain copy
     
    1. package thread;  
    2.   
    3. import java.util.concurrent.locks.Condition;  
    4. import java.util.concurrent.locks.Lock;  
    5. import java.util.concurrent.locks.ReentrantLock;  
    6. /** 
    7.  *  
    8.  * @author zhangliang 
    9.  * 
    10.  * 2016年4月8日 下午5:48:54 
    11.  */  
    12. public class ConTest {  
    13.       
    14.      final Lock lock = new ReentrantLock();  
    15.      final Condition condition = lock.newCondition();  
    16.   
    17.     public static void main(String[] args) {  
    18.         // TODO Auto-generated method stub  
    19.         ConTest test = new ConTest();  
    20.         Producer producer = test.new Producer();  
    21.         Consumer consumer = test.new Consumer();  
    22.                 
    23.           
    24.         consumer.start();   
    25.         producer.start();  
    26.     }  
    27.       
    28.      class Consumer extends Thread{  
    29.            
    30.             @Override  
    31.             public void run() {  
    32.                 consume();  
    33.             }  
    34.                 
    35.             private void consume() {  
    36.                                
    37.                     try {  
    38.                            lock.lock();  
    39.                         System.out.println("我在等一个新信号"+this.currentThread().getName());  
    40.                         condition.await();  
    41.                           
    42.                     } catch (InterruptedException e) {  
    43.                         // TODO Auto-generated catch block  
    44.                         e.printStackTrace();  
    45.                     } finally{  
    46.                         System.out.println("拿到一个信号"+this.currentThread().getName());  
    47.                         lock.unlock();  
    48.                     }  
    49.                   
    50.             }  
    51.         }  
    52.        
    53.      class Producer extends Thread{  
    54.            
    55.             @Override  
    56.             public void run() {  
    57.                 produce();  
    58.             }  
    59.                 
    60.             private void produce() {                   
    61.                     try {  
    62.                            lock.lock();  
    63.                            System.out.println("我拿到锁"+this.currentThread().getName());  
    64.                             condition.signalAll();                             
    65.                         System.out.println("我发出了一个信号："+this.currentThread().getName());  
    66.                     } finally{  
    67.                         lock.unlock();  
    68.                     }  
    69.                 }  
    70.      }  
    71.           
    72. }  

    运行结果：

    Condition的执行方式，是当在线程Consumer中调用await方法后，线程Consumer将释放锁，并且将自己沉睡，等待唤醒，线程Producer获取到锁后，开始做事，完毕后，调用Condition的signalall方法，唤醒线程Consumer，线程Consumer恢复执行。

    以上说明Condition是一个多线程间协调通信的工具类，使得某个，或者某些线程一起等待某个条件（Condition）,只有当该条件具备( signal 或者 signalAll方法被带调用)时 ，这些等待线程才会被唤醒，从而重新争夺锁。

    Condition实现生产者、消费者模式：

     

    [java] view plain copy
     
    1. package thread;  
    2.   
    3. import java.util.PriorityQueue;  
    4. import java.util.concurrent.locks.Condition;  
    5. import java.util.concurrent.locks.Lock;  
    6. import java.util.concurrent.locks.ReentrantLock;  
    7.   
    8. public class ConTest2 {  
    9.         private int queueSize = 10;  
    10.         private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);  
    11.         private Lock lock = new ReentrantLock();  
    12.         private Condition notFull = lock.newCondition();  
    13.         private Condition notEmpty = lock.newCondition();  
    14.            
    15.         public static void main(String[] args) throws InterruptedException  {  
    16.             ConTest2 test = new ConTest2();  
    17.             Producer producer = test.new Producer();  
    18.             Consumer consumer = test.new Consumer();                
    19.             producer.start();  
    20.             consumer.start();  
    21.             Thread.sleep(0);  
    22.             producer.interrupt();  
    23.             consumer.interrupt();  
    24.         }  
    25.             
    26.         class Consumer extends Thread{              
    27.             @Override  
    28.             public void run() {  
    29.                 consume();  
    30.             }  
    31.             volatile boolean flag=true;    
    32.             private void consume() {  
    33.                 while(flag){  
    34.                     lock.lock();  
    35.                     try {  
    36.                         while(queue.isEmpty()){  
    37.                             try {  
    38.                                 System.out.println("队列空，等待数据");  
    39.                                 notEmpty.await();  
    40.                             } catch (InterruptedException e) {                              
    41.                                 flag =false;  
    42.                             }  
    43.                         }  
    44.                         queue.poll();                //每次移走队首元素  
    45.                         notFull.signal();  
    46.                         System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");  
    47.                     } finally{  
    48.                         lock.unlock();  
    49.                     }  
    50.                 }  
    51.             }  
    52.         }  
    53.             
    54.         class Producer extends Thread{              
    55.             @Override  
    56.             public void run() {  
    57.                 produce();  
    58.             }  
    59.             volatile boolean flag=true;    
    60.             private void produce() {  
    61.                 while(flag){  
    62.                     lock.lock();  
    63.                     try {  
    64.                         while(queue.size() == queueSize){  
    65.                             try {  
    66.                                 System.out.println("队列满，等待有空余空间");  
    67.                                 notFull.await();  
    68.                             } catch (InterruptedException e) {  
    69.                                   
    70.                                 flag =false;  
    71.                             }  
    72.                         }  
    73.                         queue.offer(1);        //每次插入一个元素  
    74.                         notEmpty.signal();  
    75.                         System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));  
    76.                     } finally{  
    77.                         lock.unlock();  
    78.                     }  
    79.                 }  
    80.             }  
    81.         }  
    82.     }  

    运行结果如下：

    condition实现分析：

    • Condition接口包含了多种await方式和两个通知方法
    • ConditionObject实现了Condition接口，是AbstractQueuedSynchronizer的内部类
    • Reentrantlock的newCondition方法返回与某个lock实例相关的Condition对象
    [java] view plain copy
     
    1. public abstract class AbstractQueuedLongSynchronizer  
    2.     extends AbstractOwnableSynchronizer  
    3.     implements java.io.Serializable {  

    

    

    

    [java] view plain copy

     
    

    

    

      

    1. <span style="font-size:18px;">结合上面的类图，我们看到condition实现是依赖于aqs，而aqs是个抽象类。里面定义了同步器的基本框架，实现了基本的结构功能。只留有状态条件的维护由具体同步器根据具体场景来定制，如常见的 ReentrantLock 、 RetrantReadWriteLock和CountDownLatch 等等，AQS内容太多，尽量只简明梳理condition相关流程，不太深入理解底层源码。</span>  
      2. 

    

    

    下面结合上面demo来分析流程。
    

    reentrantLock.newCondition() 返回的是Condition的一个实现，该类在AbstractQueuedSynchronizer(AQS)中被实现，叫做newCondition（）
    


     
    

    

    

    [java] view plain copy

     
    

    

    

      

    1. public Condition newCondition() {  
      2. 

    2.     return sync.newCondition();  
      3. 

    3. }  
      4. 

    

    我们看一下这个await的方法，它是AQS的方法，
    


    

    


    
public final void await() throws InterruptedException {

    
    

    

    

    

    


    
02
if (Thread.interrupted())

    
    

    

    

    

    


    
03
 throw new InterruptedException();

    
    

    

    

    

    


    
04
 Node node = addConditionWaiter(); //将当前线程包装下后，

    
    

    

    

    

    


    
05
                                   //添加到Condition自己维护的一个链表中。

    
    

    

    

    

    


    
06
int savedState = fullyRelease(node);//释放当前线程占有的锁，从demo中看到，

    
    

    

    

    

    


    
07
                                       //调用await前，当前线程是占有锁的

    
    

    

    

    

    


    
08
 

    
    

    

    

    

    


    
09
int interruptMode = 0;

    
    

    

    

    

    


    
10
 while (!isOnSyncQueue(node)) {//释放完毕后，遍历AQS的队列，看当前节点是否在队列中，

    
    

    

    

    

    


    
11
                           //不在 说明它还没有竞争锁的资格，所以继续将自己沉睡。

    
    

    

    

    

    


    
12
                             //直到它被加入到队列中，聪明的你可能猜到了，

    
    

    

    

    

    


    
13
                            //没有错，在singal的时候加入不就可以了？

    
    

    

    

    

    


    
14
 LockSupport.park(this);

    
    

    

    

    

    


    
15
 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

    
    

    

    

    

    


    
16
 break;

    
    

    

    

    

    


    
17
 }

    
    

    

    

    

    


    
18
//被唤醒后，重新开始正式竞争锁，同样，如果竞争不到还是会将自己沉睡，等待唤醒重新开始竞争。

    
    

    

    

    

    


    
19
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

    
    

    

    

    

    


    
20
 interruptMode = REINTERRUPT;

    
    

    

    

    

    


    
21
 if (node.nextWaiter != null)

    
    

    

    

    

    


    
22
 unlinkCancelledWaiters();

    
    

    

    

    

    


    
23
 if (interruptMode != 0)

    
    

    

    

    

    


    
24
 reportInterruptAfterWait(interruptMode);

    
    

    

    

    

    


    
25
 }

    
    

    

    

    回到上面的demo，锁被释放后，线程Consumer开始沉睡，这个时候线程因为线程Consumer沉睡时，会唤醒AQS队列中的头结点，所所以线程Producer会开始竞争锁，并获取到，执行完后线程Producer会调用signal方法，“发出”signal信号，signal方法如下：
    


    

    


    
1
public final void signal() {

    
    

    

    

    

    


    
2
 if (!isHeldExclusively())

    
    

    

    

    

    


    
3
 throw new IllegalMonitorStateException();

    
    

    

    

    

    


    
4
 Node first = firstWaiter; //firstWaiter为condition自己维护的一个链表的头结点，

    
    

    

    

    

    


    
5
                          //取出第一个节点后开始唤醒操作

    
    

    

    

    

    


    
6
 if (first != null)

    
    

    

    

    

    


    
7
 doSignal(first);

    
    

    

    

    

    


    
8
 }

    
    

    

    

    说明下，其实Condition内部维护了等待队列的头结点和尾节点，该队列的作用是存放等待signal信号的线程，该线程被封装为Node节点后存放于此。
    


    

    而Condition自己也维护了一个队列，该队列的作用是维护一个等待signal信号的队列，两个队列的作用是不同，事实上，每个线程也仅仅会同时存在以上两个队列中的一个，流程是这样的：
    

    


    注意：
    

    1.线程producer调用signal方法，这个时候Condition的等待队列中只有线程Consumer一个节点，于是它被取出来，并被加入到AQS的等待队列中。  注意，这个时候，线程Consumer 并没有被唤醒。
    

    2.Sync是AQS的抽象子类，实现可重入和互斥的大部分功能。在Sync的子类中有FairSync和NonfairSync两种代表公平锁策略和非公平锁策略。Sync lock方法留给子类去实现，NonfairSync的实现：
    


    

    

    [java] view plain copy

     
    


    


    

      

    1. final void lock() {  
      2. 

    2.            if (compareAndSetState(0, 1))  
      3. 

    3.                setExclusiveOwnerThread(Thread.currentThread());  
      4. 

    4.            else  
      5. 

    5.                acquire(1);  
      6. 

    6.        }  
      7. 


    

    
其中如果一开始获取锁成功，是直接设置当前线程。
    


    否则执行acquire(1),也就是进入aqs等待队列。这里不展开细节。
    

    可以这样理解，整个协作过程是靠结点在AQS的等待队列和Condition的等待队列中来回移动实现的，每个队列的意义不同，Condition作为一个条件类，很好的自己维护了一个等待信号的队列，并在适时的时候将结点加入到AQS的等待队列中来实现的唤醒操作
    

    
    
                
    
                
              
    • 
              
    
              
  • 

                相关阅读:
    
                
                  flink 使用processFunction函数的sideOutPut实现filter操作(java版)
                    
    flink 状态编程之RichFlatMapFunction用法-java
                    
    java flink之eventTime和window
                    
    java spark sql 计算各个省份广告点击数的top3 
                    
    java spark 计算各个省份广告点击数的top3
                    
    java streaming转换算子window,countByWindow
                    
    java spark转换算子sortByKey
                    
    java spark转换算子join、leftOuterJoin、rightOuterJoin、fullOuterjoin
                    
    java  spark转换算子groupByKey、reduceByKey、aggregateByKey
                    
    java spark转换算子union、intersection、subtract
                    
                        
              
    • 
              
    
              
  • 
                原文地址：https://www.cnblogs.com/chenshaogang/p/8849807.html
              
    • 

            

          



          
          

            

            

          

        

      

      



      

      

        

          Copyright © 2011-2022 走看看