xylz,imxylz

关注后端架构、中间件、分布式和并发编程

   :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理 ::
  111 随笔 :: 10 文章 :: 2680 评论 :: 0 Trackbacks

本小节介绍锁释放Lock.unlock()。

Release/TryRelease

unlock操作实际上就调用了AQS的release操作,释放持有的锁。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

前面提到过tryRelease(arg)操作,此操作里面总是尝试去释放锁,如果成功,说明锁确实被当前线程持有,那么就看AQS队列中的头结点是否为空并且能否被唤醒,如果可以的话就唤醒继任节点(下一个非CANCELLED节点,下面会具体分析)。

对于独占锁而言,java.util.concurrent.locks.ReentrantLock.Sync.tryRelease(int)展示了如何尝试释放锁(tryRelease)操作。

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

整个tryRelease操作是这样的:

    1. 判断持有锁的线程是否是当前线程,如果不是就抛出IllegalMonitorStateExeception(),因为一个线程是不能释放另一个线程持有的锁(否则锁就失去了意义)。否则进行2。
    2. 将AQS状态位减少要释放的次数(对于独占锁而言总是1),如果剩余的状态位0(也就是没有线程持有锁),那么当前线程就是最后一个持有锁的线程,清空AQS持有锁的独占线程。进行3。
    3. 将剩余的状态位写回AQS,如果没有线程持有锁就返回true,否则就是false。

参考上一节的分析就可以知道,这里c==0决定了是否完全释放了锁。由于ReentrantLock是可重入锁,因此同一个线程可能多重持有锁,那么当且仅当最后一个持有锁的线程释放锁是才能将AQS中持有锁的独占线程清空,这样接下来的操作才需要唤醒下一个需要锁的AQS节点(Node),否则就只是减少锁持有的计数器,并不能改变其他操作。

tryRelease操作成功后(也就是完全释放了锁),release操作才能检查是否需要唤醒下一个继任节点。这里的前提是AQS队列的头结点需要锁(waitStatus!=0),如果头结点需要锁,就开始检测下一个继任节点是否需要锁操作。

在上一节中说道acquireQueued操作完成后(拿到了锁),会将当前持有锁的节点设为头结点,所以一旦头结点释放锁,那么就需要寻找头结点的下一个需要锁的继任节点,并唤醒它。

private void unparkSuccessor(Node node) {
        //此时node是需要是需要释放锁的头结点

        //清空头结点的waitStatus,也就是不再需要锁了
        compareAndSetWaitStatus(node, Node.SIGNAL, 0);

        //从头结点的下一个节点开始寻找继任节点,当且仅当继任节点的waitStatus<=0才是有效继任节点,否则将这些waitStatus>0(也就是CANCELLED的节点)从AQS队列中剔除  
       //这里并没有从head->tail开始寻找,而是从tail->head寻找最后一个有效节点。
       //解释在这里 http://www.blogjava.net/xylz/archive/2010/07/08/325540.html#377512

        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }

        //如果找到一个有效的继任节点,就唤醒此节点线程
        if (s != null)
            LockSupport.unpark(s.thread);
    }

这里再一次把acquireQueued的过程找出来。对比unparkSuccessor,一旦头节点的继任节点被唤醒,那么继任节点就会尝试去获取锁(在acquireQueued中node就是有效的继任节点,p就是唤醒它的头结点),如果成功就会将头结点设置为自身,并且将头结点的前任节点清空,这样前任节点(已经过时了)就可以被GC释放了。

final boolean acquireQueued(final Node node, int arg) {
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } catch (RuntimeException ex) {
        cancelAcquire(node);
        throw ex;
    }
}

setHead中,将头结点的前任节点清空并且将头结点的线程清空就是为了更好的GC,防止内存泄露。

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

对比lock()操作,unlock()操作还是比较简单的,主要就是释放响应的资源,并且唤醒AQS队列中有效的继任节点。这样所就按照请求的顺序去尝试获取锁了。

整个lock()/unlock()过程完成了,我们再回头看公平锁(FairSync)和非公平锁(NonfairSync)。

公平锁和非公平锁只是在获取锁的时候有差别,其它都是一样的。

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

在上面非公平锁的代码中总是优先尝试当前是否有线程持有锁,一旦没有任何线程持有锁,那么非公平锁就霸道的尝试将锁“占为己有”。如果在抢占锁的时候失败就和公平锁一样老老实实的去排队。

也即是说公平锁和非公平锁只是在入AQSCLH队列之前有所差别,一旦进入了队列,所有线程都是按照队列中先来后到的顺序请求锁。

Condition

条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。

条件(也称为条件队列条件变量)为线程提供了一个含义,以便在某个状态条件现在可能为 true 的另一个线程通知它之前,一直挂起该线程(即让其“等待”)。因为访问此共享状态信息发生在不同的线程中,所以它必须受保护,因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是:以原子方式 释放相关的锁,并挂起当前线程,就像 Object.wait 做的那样。

上述API说明表明条件变量需要与锁绑定,而且多个Condition需要绑定到同一锁上。前面的Lock中提到,获取一个条件变量的方法是Lock.newCondition()

void await() throws InterruptedException;
void awaitUninterruptibly();
long awaitNanos(long nanosTimeout) throws InterruptedException;
boolean await(long time, TimeUnit unit) throws InterruptedException;
boolean awaitUntil(Date deadline) throws InterruptedException;
void signal();
void signalAll();

以上是Condition接口定义的方法,await*对应于Object.waitsignal对应于Object.notifysignalAll对应于Object.notifyAll。特别说明的是Condition的接口改变名称就是为了避免与Object中的wait/notify/notifyAll的语义和使用上混淆,因为Condition同样有wait/notify/notifyAll方法。

每一个Lock可以有任意数据的Condition对象,Condition是与Lock绑定的,所以就有Lock的公平性特性:如果是公平锁,线程为按照FIFO的顺序从Condition.await中释放,如果是非公平锁,那么后续的锁竞争就不保证FIFO顺序了。

一个使用Condition实现生产者消费者的模型例子如下。

package xylz.study.concurrency.lock;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ProductQueue<T> {

    private final T[] items;

    private final Lock lock = new ReentrantLock();

    private Condition notFull = lock.newCondition();

    private Condition notEmpty = lock.newCondition();

    //
    private int head, tail, count;

    public ProductQueue(int maxSize) {
        items = (T[]) new Object[maxSize];
    }

    public ProductQueue() {
        this(10);
    }

    public void put(T t) throws InterruptedException {
        lock.lock();
        try {
            while (count == getCapacity()) {
                notFull.await();
            }
            items[tail] = t;
            if (++tail == getCapacity()) {
                tail = 0;
            }
            ++count;
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            T ret = items[head];
            items[head] = null;//GC
            //
            if (++head == getCapacity()) {
                head = 0;
            }
            --count;
            notFull.signalAll();
            return ret;
        } finally {
            lock.unlock();
        }
    }

    public int getCapacity() {
        return items.length;
    }

    public int size() {
        lock.lock();
        try {
            return count;
        } finally {
            lock.unlock();
        }
    }

}

在这个例子中消费take()需要 队列不为空,如果为空就挂起(await()),直到收到notEmpty的信号;生产put()需要队列不满,如果满了就挂起(await()),直到收到notFull的信号。

可能有人会问题,如果一个线程lock()对象后被挂起还没有unlock,那么另外一个线程就拿不到锁了(lock()操作会挂起),那么就无法通知(notify)前一个线程,这样岂不是“死锁”了?

 

await* 操作

上一节中说过多次ReentrantLock是独占锁,一个线程拿到锁后如果不释放,那么另外一个线程肯定是拿不到锁,所以在lock.lock()lock.unlock()之间可能有一次释放锁的操作(同样也必然还有一次获取锁的操作)。我们再回头看代码,不管take()还是put(),在进入lock.lock()后唯一可能释放锁的操作就是await()了。也就是说await()操作实际上就是释放锁,然后挂起线程,一旦条件满足就被唤醒,再次获取锁!

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

上面是await()的代码片段。上一节中说过,AQS在获取锁的时候需要有一个CHL的FIFO队列,所以对于一个Condition.await()而言,如果释放了锁,要想再一次获取锁那么就需要进入队列,等待被通知获取锁。完整的await()操作是安装如下步骤进行的:

    1. 将当前线程加入Condition锁队列。特别说明的是,这里不同于AQS的队列,这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。
    2. 释放锁。这里可以看到将锁释放了,否则别的线程就无法拿到锁而发生死锁。进行3。
    3. 自旋(while)挂起,直到被唤醒或者超时或者CACELLED等。进行4。
    4. 获取锁(acquireQueued)。并将自己从Condition的FIFO队列中释放,表明自己不再需要锁(我已经拿到锁了)。

这里再回头介绍Condition的数据结构。我们知道一个Condition可以在多个地方被await*(),那么就需要一个FIFO的结构将这些Condition串联起来,然后根据需要唤醒一个或者多个(通常是所有)。所以在Condition内部就需要一个FIFO的队列。

private transient Node firstWaiter;
private transient Node lastWaiter;

上面的两个节点就是描述一个FIFO的队列。我们再结合前面提到的节点(Node)数据结构。我们就发现Node.nextWaiter就派上用场了!nextWaiter就是将一系列的Condition.await*串联起来组成一个FIFO的队列。

 

signal/signalAll 操作

await*()清楚了,现在再来看signal/signalAll就容易多了。按照signal/signalAll的需求,就是要将Condition.await*()中FIFO队列中第一个Node唤醒(或者全部Node)唤醒。尽管所有Node可能都被唤醒,但是要知道的是仍然只有一个线程能够拿到锁,其它没有拿到锁的线程仍然需要自旋等待,就上上面提到的第4步(acquireQueued)。

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

private void doSignalAll(Node first) {
    lastWaiter = firstWaiter  = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}

上面的代码很容易看出来,signal就是唤醒Condition队列中的第一个非CANCELLED节点线程,而signalAll就是唤醒所有非CANCELLED节点线程。当然了遇到CANCELLED线程就需要将其从FIFO队列中剔除。

final boolean transferForSignal(Node node) {
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    Node p = enq(node);
    int c = p.waitStatus;
    if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

上面就是唤醒一个await*()线程的过程,根据前面的小节介绍的,如果要unpark线程,并使线程拿到锁,那么就需要线程节点进入AQS的队列。所以可以看到在LockSupport.unpark之前调用了enq(node)操作,将当前节点加入到AQS队列。

整个锁机制的原理就介绍完了,从下一节开始就进入了锁机制的应用了。

 



©2009-2014 IMXYLZ |求贤若渴
posted on 2010-07-08 12:33 imxylz 阅读(30398) 评论(11)  编辑  收藏 所属分类: J2EE

评论

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2010-07-08 12:35 草包书生
还可以.......  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2012-05-07 10:12 红泪
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
有个问题想请教一下您,就是这段代码中,为什么要从tail往前遍历寻找有效继任节点呢?CHL队列不是一个先进先出的双环队列么?按照这个的话,我觉得应该是直接从head往后遍历寻找有效继任节点才对?

希望博主能够给我分析分析这个问题,万分感谢!  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2012-05-07 11:36 imxylz
@红泪
理论上讲如果在释放节点的时候其他所有节点都没有被中断(也就是节点没有被CANCELLED),那么就应当唤醒头节点的下一个有效节点(头节点是傀儡节点),也就是从head往后寻找有效继任节点。

但是我们知道所有调用了lock()/tryLock(long time, TimeUnit unit)的线程可能会被中断,这时候已经进入CHL队列的节点node就会被CANCELLED,也就是会移出队列。
而移出队列的逻辑有点复杂,有空我单独写一篇文章。
简单说就是用被移出节点node的继任节点next替换前任有效节点的next。
用代码描述就是java.util.concurrent.locks.AbstractQueuedSynchronizer.cancelAcquire(Node):
node=cancelled_node;
cas(node.pre.next,node,node.next);
并且将node.next指向node,也就是node没有继任节点了,但是不修改前任节点。
也就是说如果从后tail往前遍历到被删出节点node时,根据node.pre可以继续往前移动,直到移动到head为止。

如果要想从head往后遍历,那么代码逻辑就是:
node = cancelled_node;
cas(node.next.pre,node,node.pre);

这两处的逻辑差别在于,由于存在一个傀儡节点(head),因此节点node.pre总是存在的,处理起来稍微容易点。  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2012-05-08 10:25 imxylz
@imxylz
@红泪
上面的回答可能不够正确 有空我在写一篇文章吧  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2012-05-12 11:44 红泪
呵呵.谢谢博主的解答!!我觉得谁要有空结合Condition将chl队列的变化情况写写,这个挺好..不过我觉得这个还是挺复杂,功力不足唉.  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2012-07-21 23:40 ToO
"在上一节中说道acquireQueued操作完成后(拿到了锁),会将当前持有锁的节点设为头结点",也即是此时head 指向的是当前要释放锁的节点,所以head已经不在是傀儡节点了  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2013-03-25 01:00 ffengtian
@imxylz
从队尾获取可用的Node,执行解锁后,试图获取锁时,它的pred没有指向头节点造成重新阻塞  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2014-12-01 15:56 会飞的鸭子
博主关于您写的那个队列,head做自加操作,是否需要判断head跟tail的关系呢?如果head跟tail已经指向同一节点,那么此时做自加操作会不会有问题呢?
if (++head == getCapacity()) {
head = 0;
}
改为
if(head!=tail){
head++;
if(head==getCapacity()){
head=0;
}
}
  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2015-06-28 16:02 mitisky
请教一个问题:线程1,2.同时去lock,假设线程1获得锁,线程2加到等待队列。如果在线程2加入等待队列之前,线程1就unlock。也就是说线程2在加入等待队列的前一步,线程1完成了unlock操作。那这是不是意味着线程2会等待,线程1的unlock并没有成功唤醒线程2  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2015-06-29 09:25 imxylz
@mitisky
入队列enq 是一个CAS操作,如果加入队列成功,会立即进行队列自旋操作,在这个操作里面尝试获取锁。
参考:http://www.blogjava.net/xylz/archive/2010/07/07/325410.html

acquireQueued(node,arg)
  回复  更多评论
  

# re: 深入浅出 Java Concurrency (9): 锁机制 part 4 2016-07-11 16:17 guanzhisong
@imxylz
我觉得是因为被cancelled的节点,next指针会指向自己
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
unparkSuccessor(node);
}

node.next = node; // help GC
这段代码是cancelAcquire里面的,假如通过next指针去遍历,链表其实已经被中断了,但是链表的prev指针貌似就没有这个问题,个人愚见  回复  更多评论
  


只有注册用户登录后才能发表评论。


网站导航:
 

©2009-2014 IMXYLZ