ReentrantLock和AQS

亦凉 2022-05-15 08:00 246阅读 0赞

demo

public static void main(String[]args){ 
    		final ReentrantLock lock = new ReentrantLock();
    		for(int j=0;j<5;j++){ 
    			new Thread(new Runnable(){ 
    				public void run(){ 
    					try{ 
    					   //加锁
    						lock.lock();  
    						for (int i = 0; i < 5; i++) { 
    							System.out.println(Thread.currentThread().getName()+" : "+i);
    						}
    					}catch(Exception e){ 
    						e.printStackTrace();
    					}finally{ 
    					   //释放锁
    						lock.unlock();
    					}
    				}
    			}).start();
    		}
    	}

lock.lock(); 等同synchronized，  
lock.unlock(); 等同synchronized的方法块结束

public class ReentrantLock implements Lock, java.io.Serializable { 
     private final Sync sync;
        
        //默认是非公平锁
        public ReentrantLock() { 
            sync = new NonfairSync();
        }
        
        /** *下面是经常用到的2个方法，lock 和 unlock *都用到AQS里的方法 **/
        
        
        //!!!最重要的方法-lock
        public void lock() { 
            sync.lock();
        }
        
        //!!!最重要的方法-unlock
        public void unlock() { 
            sync.release(1);
        }
        
    
        //!!!重点 AbstractQueuedSynchronizer 简称为AQS 
        //ReentrantLock里的抽象类Sync实现AbstractQueuedSynchronizer
        //抽象类Sync又衍生出2个实现类，公平锁和非公平锁
        abstract static class Sync extends AbstractQueuedSynchronizer { 
            //留了一个lock的抽象方法
            abstract void lock();
        
            //默认实现了非公平锁的获取
            final boolean nonfairTryAcquire(int acquires) { 
                final Thread current = Thread.currentThread();
                //volatile实现了可见性
                int c = getState();
                if (c == 0) { 
                    //unsafe里的cas写，区别于公平锁的地方
                    //队列里的线程和这里的线程会共同去竞争锁
                    //这对于排队的线程是不公平的
                    if (compareAndSetState(0, acquires)) { 
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) { 
                    int nextc = c + acquires;
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
        }
        
        //公平锁
         static final class FairSync extends Sync { 
            final void lock() { 
                acquire(1);
            }
          protected final boolean tryAcquire(int acquires) { 
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) { 
                   //公平锁区分非公平锁就在!hasQueuedPredecessors()上
                   //如果判断有线程在队列里排队，当前请求线程乖乖去排队吧
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) { 
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) { 
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
        }
    
        }
        
       //非公平锁 
         static final class NonfairSync extends Sync { 
            private static final long serialVersionUID = 7316153563782823691L;
            final void lock() { 
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) { 
                return nonfairTryAcquire(acquires);
            }
        }

下面列出了公平锁和非公平锁的区别

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>公平锁</th> 
   <th>非公平锁</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>lock()</td> 
   <td>一进来就参与竞争锁</td> 
   <td>不参与竞争锁</td> 
  </tr> 
  <tr> 
   <td>tryAcquire()</td> 
   <td>不判断是否有线程在排队等待获取锁</td> 
   <td>判断是否有线程在排队等待获取锁</td> 
  </tr> 
 </tbody> 
</table>

ReentrantLock默认使用非公平锁

static final class NonfairSync extends Sync { 
            private static final long serialVersionUID = 7316153563782823691L;
            
            
            final void lock() { 
               //cas 更新AQS里的state 从 0->1 
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                //cas 更新不成功，调用AQS里的acquire
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) { 
                return nonfairTryAcquire(acquires);
            }
        }

AbstractQueuedSynchronizer(AQS)里的acquire

public final void acquire(int arg) { 
            //tryAcquire先获取一次，
            //如果失败则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            //addWaiter(Node.EXCLUSIVE)是将当前的线程封装成为一个node，放到等待队列的末尾
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

addWaiter(Node.EXCLUSIVE)

private Node addWaiter(Node mode) { 
            Node node = new Node(Thread.currentThread(), mode);
           //如果尾节点不为空的话，就将自己添加到链表的尾部
            Node pred = tail;
            if (pred != null) { 
                node.prev = pred;
                if (compareAndSetTail(pred, node)) { 
                    pred.next = node;
                    return node;
                }
            }
            //如果上面那步不成功的话，就调用enq(node)
            //其实里面的逻辑和上面差不多，
            //多了尾节点为空判断
            //然后循环上面的过程直到加入成功
            enq(node);
            return node;
        }

AQS里的队列结构,上面新加的节点将会放到下面双向链表的结尾，  
加入成功后，调用acquireQueued去竞争锁

final boolean acquireQueued(final Node node, int arg) { 
            boolean failed = true;
            try { 
                boolean interrupted = false;
                for (;;) { 
                    final Node p = node.predecessor();
                    //在循环中尝试去将AQS里的state将0->1
                    if (p == head && tryAcquire(arg)) { 
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    //如果尝试2次获取失败，则会调用lockSupport的挂起操作，
                    //挂起线程
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }

挂起的线程等待lock.unlock()唤醒,unlock调用sync.release(1)

public final boolean release(int arg) { 
        //将AQS里的state -1，如果state减到0
            if (tryRelease(arg)) { 
                Node h = head;
                //去唤醒队列中的后继的那个节点
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

tryRelease是个抽象方法，ReentrantLock的syn内部类里实现

protected final boolean tryRelease(int releases) { 
    //将AQS里的state -1
                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;
            }

unparkSuccessor

private void unparkSuccessor(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. */
            int ws = node.waitStatus;
            if (ws < 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. 上面的解释是如果后续节点是正常的,即node.waiteStauts=0, 那就唤醒这个节点，如果node.waiteStauts>0,就是这个thread被cancel了， 比如lock.tryAcquireNanos(int arg, long nanosTimeout)，如果获取锁超时 则表示这个node标记为cancel, 如果节点是cancel,倒序寻找到未被标记为cancelled的节点 */
            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)
               //唤醒节点，唤醒的节点重新进入竞争锁（将state 0->1）
                LockSupport.unpark(s.thread);
        }

> 简单的描述上面的过程，如果多个线程同时去 调用lock.lock() 即将AQS里的state从0改成1，这个更改是通过CAS 来更改的。也就是只有一个线程会更改成功，更改成功的这个线程就获得了锁。其它线程则会排队，  
> 排完队后，线程会去尝试获取锁，尝试2次，如果失败了，则线程挂起。 等待lock.unlock()来唤醒。

--------------------

重新复习了下AQS. 对AQS里的锁的机制又有个新的认识。  
该图来之[这位的博客][Link 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]

> 将node连在一起形成了单链表不是特别好，事实上代码里并没有形成单链表的数据结构。逻辑上可以这么认为。

public class CLHLockDemo { 
    
        public static void main(String[] args) { 
           
            CLHLock lock =new CLHLock();
            for(int i=0;i<10;i++){ 
                new Thread(()->{ 
                    lock.lock(); //加锁
                    System.out.println(Thread.currentThread().getName()+" is working ");
                    try { 
                        sleep(1000);
                        System.out.println(Thread.currentThread().getName()+" end working ");
                        lock.unlock(); //解锁
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }).start();
            }
        }
        static class CLHLock{ 
            private  ThreadLocal<CLHNode> curNodeInThread;
            private AtomicReference<CLHNode> tail;
            public CLHLock(){ 
                curNodeInThread=new ThreadLocal();
                tail=new AtomicReference();
            }
            public void lock(){ 
                CLHNode curNode=new CLHNode();
                curNodeInThread.set(curNode);
                CLHNode preNode=tail.getAndSet(curNode);
                if(preNode!=null){ 
                    while(preNode.isWork()){ 
                        //spin here
                    }
                }
            }
            public void unlock(){ 
                CLHNode curNode=curNodeInThread.get();
                curNodeInThread.remove();
                if(curNode!=null&&curNode.isWork()){ 
                    curNode.setWork(false);
                }
            }
        }
        static class CLHNode{ 
            private volatile boolean work=true;
    
            public boolean isWork() { 
                return work;
            }
    
            public void setWork(boolean work) { 
                this.work = work;
            }
        }
    }

[Link 1]: https://blog.csdn.net/claram/article/details/83828768
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]: /images/20220515/e0ebb24860ae4425ad13420a20c645f8.png