Java并发：Condition

港控/mmm° 2022-12-01 11:58 221阅读 0赞

Condition使用案例：消费模型

public class ConditionTest { 
        public static ReentrantLock lock = new ReentrantLock();
        public static Condition notEmpty = lock.newCondition();
        public static Condition notFull = lock.newCondition();
        public static PriorityQueue<Integer> queue = new PriorityQueue<Integer>(10);
    
        public static void main(String[] args) throws Exception{ 
            // 消费线程
            new Thread(()->{ 
                while (true) { 
                    lock.lock();
                    while (queue.isEmpty()){ 
                        try { 
                            System.out.println("队列为空，等待数据");
                            notEmpty.await();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                    queue.poll();
                    System.out.println("移走一个元素，队列中还剩下"+queue.size()+"个元素");
                    notFull.signal();
                    lock.unlock();
                }
            }).start();
    
            // 入队线程
            new Thread(()->{ 
                while (true){ 
                    lock.lock();
                    while (queue.size()==10){ 
                        System.out.println("队列满了，等待消费");
                        try { 
                            notFull.await();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                    queue.offer(1);
                    notEmpty.signal();
                    System.out.println("向队列中插入一个元素，队列中元素为"+queue.size()+"个元素");
                    lock.unlock();
                }
            }).start();
        }
    
    }

**Condition类结构：**  
ConditionObject是AQS的内部类，类中维护一条Node节点的队列

public class ConditionObject implements Condition, java.io.Serializable { 
            // Condition队列的头结点
            private transient Node firstWaiter;
            // Condition队列的尾节点
            private transient Node lastWaiter;
            // 构造方法
            public ConditionObject() {  }
      }

Condition的主要方法为await()，singal()，我们就主要看一下这两个方法  
**await:**

public final void await() throws InterruptedException { 
    			// 线程如果有中断标记，则抛出异常
                if (Thread.interrupted())
                    throw new InterruptedException();
                // 添加节点到Condition队列中 
                Node node = addConditionWaiter();
                // 释放当前持有线程，saveState为释放锁之后的状态
                int savedState = fullyRelease(node);
                int interruptMode = 0;
                // 判断节点是否在AQS队列中
                while (!isOnSyncQueue(node)) { 
                	// 挂起线程
                	// await方法在这里线程挂起了
                	// 等待线程节点加入到AQS队列中，知道被唤醒
                    LockSupport.park(this);
                    // 没有中断标记则返回0
                    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                        break;
                }
                //被唤醒后，重新开始正式竞争锁，同样，如果竞争不到还是会将自己沉睡，等待唤醒重新开始竞争。
                if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                    interruptMode = REINTERRUPT;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
            }
            // 添加节点到等待队列
            private Node addConditionWaiter() { 
            	// 指向尾节点
                Node t = lastWaiter;
                // 如果尾节点不为null，且尾节点的状态为CANCELLED，则移除
                if (t != null && t.waitStatus != Node.CONDITION) { 
                    unlinkCancelledWaiters();
                    t = lastWaiter;
                }
                // 构建新节点
                Node node = new Node(Thread.currentThread(), Node.CONDITION);
                // 入队操作
                if (t == null)
                    firstWaiter = node;
                else
                    t.nextWaiter = node;
                lastWaiter = node;
                return node;
            }
            
            final int fullyRelease(Node node) { 
            boolean failed = true;
            try { 
            	// 获取状态
                int savedState = getState();
                // 释放持有的锁
                // 释放锁成功，唤醒AQS队列的第一个线程
                if (release(savedState)) { 
                    failed = false;
                    return savedState;
                } else { 
                    throw new IllegalMonitorStateException();
                }
            } finally { 
            	// 释放锁失败
                if (failed)
                    node.waitStatus = Node.CANCELLED;
            }
        }
        // 参数; Node node = new Node(Thread.currentThread(), Node.CONDITION) 
        final boolean isOnSyncQueue(Node node) { 
        	// 无前驱节点则不在AQS队列中
            if (node.waitStatus == Node.CONDITION || node.prev == null)
                return false;
            // 有后继节点则在AQS队列中 
            if (node.next != null) 
                return true;
    
            return findNodeFromTail(node);
        }
        // 从AQS队列的尾节点查找节点
        private boolean findNodeFromTail(Node node) { 
            Node t = tail;
            for (;;) { 
                if (t == node)
                    return true;
                if (t == null)
                    return false;
                t = t.prev;
            }
        }

**singal：**

// 转运一个节点到AQS队列中去
            public final void signal() { 
            	// 判断当前线程
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                // 指向Condition队列的头结点 
                Node first = firstWaiter;
                if (first != null)
                    doSignal(first);  
            }
            // 参数：队列头结点
            private void doSignal(Node first) { 
                do { 
                	// firstWaiter指向后继节点
                    if ( (firstWaiter = first.nextWaiter) == null)
                        lastWaiter = null;
                    // 
                    first.nextWaiter = null;
                } while (!transferForSignal(first) &&
                		  // 指向下一个Condition队列的下一个节点
                         (first = firstWaiter) != null);
            }
            // 
    		final boolean transferForSignal(Node node) { 
    		// 设置waitStatus从CONDITION到0，失败则返回false
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
            // 入队AQS 
            Node p = enq(node);
           	// p为AQS队列添加节点前的尾节点
            int ws = p.waitStatus;
            // 状态为CANCELLED 或者 cas 修改p节点的状态失败
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            	// 唤醒该节点
                LockSupport.unpark(node.thread);
            return true;
        }

**流程图：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/045ca2a3ca814428b98bcc7ad8b5981a.png