您现在的位置是：首页 > Java

当前栏目

Java并发：Condition

JAVA 并发状态队列线程队列

2023-03-02 11:11:39 时间

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

猜你喜欢

Android Material Design控件使用（三）——CardView 卡片布局和SnackBar使用
Java 学习笔记网络编程使用Socket传输文件 CS模式
Java 学习笔记使用并发包ReentrantLock简化生产者消费者模式代码
Java 学习笔记使用synchronized实现生产者消费者模式
Java 学习笔记观察者模式与适配者模式监控文件夹
Java 学习笔记线程控制
Acer宏碁笔记本触摸板失效解决方法
java学习笔记线程的实现与同步
Android开发过程中的坑及解决方法收录（六）
杂牌机搞机之旅（一）——获得root权限（刷入magisk）
Java 学习笔记泛型
Java 学习笔记反射与迭代器
Java 学习笔记正则表达式
Java 学习笔记执行外部命令包装类枚举类型
IDEA设置显示中文文档API方法说明
算法学习——二分查找（折半查找）
IDEA插件（Android Studio插件）开发示例代码及bug解决
Android Studio（IDEA）快速代码模版使用
IDEA解决crtl+space与搜狗输入法冲突
windows10系统盘瘦身

zl程序教程

当前栏目

Java并发：Condition

相关文章