private Node addWaiter(Node mode) { Nodenode=newNode(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Nodepred= tail; //创建新的节点,并将线程和节点关联。 //将同步队列的尾节点后继节点指向新节点, //将新节点的前驱节点指向尾节点, //新节点称为同步队列的尾节点。 if (pred != null) { node.prev = pred; //CAS操作将新节点插入到,成功则返回,不成功则继续下面的enq方法, //进行死循环CAS插入,直到成功。 if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } //如果上面的CAS操作插入不成功,则调用enq方法 死循环插入 直到成功。 enq(node); return node; }
private Node enq(final Node node) { //死循环 直到插入成功。 for (;;) { Nodet= tail; //如果尾节点为null,说明同步队列还未初始化,则CAS操作新建头节点 if (t == null) { // Must initialize if (compareAndSetHead(newNode())) tail = head; } else { //通过CAS操作将节点插入到同步队列尾部 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
//节点以“死循环”的方式去获取资源,为什么死循环加了双引号呢?因为循环并不 //是一直让节点无间断的去获取资源,节点会经历 获取资源->失败->线程进入等待 //状态->唤醒->获取资源......,线程在死循环的过程会不断等待和唤醒,即节点的自旋。 finalbooleanacquireQueued(final Node node, int arg) { booleanfailed=true; try { booleaninterrupted=false; for (;;) { //获取节点的前驱节点 finalNodep= node.predecessor(); //如果前驱节点为头结点,表示资源正在前驱节点的手上,那该节点 //去尝试获取资源,如果获取成功则将该节点设置为头结点,并且 //返回。 //如果该节点的前驱节点并不是头节点或者是前驱节点是头结点但是 //该节点获取资源失败,继续往下执行。 if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } //调用shouldParkAfterFailedAcquire函数,将该节点的前驱节点 //的状态设置为SIGNAL,告诉前驱节点我要去“睡觉”了,当资源排 //到你的时候,你就通知我一下让我醒来,即节点做进入等待状态 //的准备。 //当节点做好了进入等待状态的准备,则调用parkAndCheckInterrupt //函数,让该节点进入到等待状态。 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
privatestaticbooleanshouldParkAfterFailedAcquire(Node pred, Node node) { //获取前驱节点的状态。 intws= pred.waitStatus; //如果前驱节点的状态已经为SIGNAL了,即已经做好准备了,那直接返回。 if (ws == Node.SIGNAL) /* * This node has already set status asking a release * to signal it, so it can safely park. */ returntrue; //如果前驱节点的状态为取消状态,则将前驱节点移除队列,循环这个过程 //直到前驱节点不为取消状态为止。 if (ws > 0) { /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; //如果前驱节点没有做好准备(标志状态为SIGNAL)、前驱节点也没有被取消, //则使用CAS操作将前驱节点的状态更新为SIGNAL,然后返回false,为什么 //是返回false呢?因为CAS操作并不保证一定能更新成功,返回false的目的 //是让acquireQueued函数再执行一次for循环,这个循环第一可以让该节点 //再尝试获取资源(万一成功了呢 是吧),第二是让acquireQueued函数再调用 //一次shouldParkAfterFailedAcquire函数(即本函数)判断节点的前驱节点是 //否已经设置为SIGNAL状态了。 } else { /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } returnfalse; }
privatevoidunparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ //获取头结点状态。 intws= node.waitStatus; //如果状态小于0,即代表有后继节点需要唤醒。 if (ws < 0) //将头结点的状态置为0 因为只需要唤醒一次 compareAndSetWaitStatus(node, ws, 0);
/* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ Nodes= node.next; //如果头结点的后继节点为空 或者 头结点的后继节点处于取消状态,则从尾部开始往前寻找, //找到一个离头结点最近 且状态不是取消状态的节点。 if (s == null || s.waitStatus > 0) { s = null; for (Nodet= tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } //如果头结点的后继节点不为取消状态,则直接将后继节点唤醒 if (s != null) LockSupport.unpark(s.thread); }
privatevoidsetHeadAndPropagate(Node node, int propagate) { Nodeh= head; // Record old head for check below setHead(node); /* * Try to signal next queued node if: * Propagation was indicated by caller, * or was recorded (as h.waitStatus either before * or after setHead) by a previous operation * (note: this uses sign-check of waitStatus because * PROPAGATE status may transition to SIGNAL.) * and * The next node is waiting in shared mode, * or we don't know, because it appears null * * The conservatism in both of these checks may cause * unnecessary wake-ups, but only when there are multiple * racing acquires/releases, so most need signals now or soon * anyway. */ if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Nodes= node.next; //如果节点为共享节点,则调用doReleaseShared函数唤醒后继节点。 if (s == null || s.isShared()) doReleaseShared(); } }
privatevoiddoReleaseShared() { /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. */ for (;;) { Nodeh= head; if (h != null && h != tail) { intws= h.waitStatus; //如果节点标识后继节点需要唤醒,则调用unparkSuccessor方法进行唤醒。 if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } elseif (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } }