AQS源码分析之中断与超时获取锁

￡神魔★判官ぃ 2022-12-10 01:19 204阅读 0赞

# AQS源码分析之中断与超时获取锁 #

## 中断基础 ##

*  Thread.interrupt()：对象方法，置中断标记为true。
 *  Thread.currentThread().isInterrupted()：对象方法，返回线程当前的中断标记状态，不会清除中断标志位。
 *  Thread.interrupted()：静态方法，返回线程当前的中断标记状态同时清除中断标志位（置为false）。

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.LockSupport;
    
    public class InterruptDemo { 
        public static void main(String[] args) throws InterruptedException { 
    
            Thread t1 = new Thread(() -> { 
                LockSupport.park();
                System.out.println("t1:" + Thread.interrupted()); // true
                System.out.println("t1:" + Thread.currentThread().isInterrupted()); // false
            });
            t1.start();
    
            TimeUnit.SECONDS.sleep(1);
            t1.interrupt();
            System.out.println("main:" + t1.isInterrupted()); // true
        }
    }

## 中断方法lockInterruptibly()的源码分析 ##

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 演示独占锁的中断 */
    public class InterruptThreadDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            ReentrantLock reentrantLock = new ReentrantLock();
            new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    System.out.println("t1");
                    TimeUnit.SECONDS.sleep(30);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t1").start();
    
            TimeUnit.SECONDS.sleep(1); // 等待t1启动
    
            Thread t2 = new Thread(() -> { 
                try { 
                    reentrantLock.lockInterruptibly();
                    try { 
                        System.out.println("t2 get lock");
                    } finally { 
                        reentrantLock.unlock();
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }, "t2");
            t2.start();
    
            TimeUnit.SECONDS.sleep(1); // 等待t2启动
    
            t2.interrupt(); // 中断t2，会抛出InterruptedException异常
        }
    
    }

java.util.concurrent.locks.ReentrantLock\#lockInterruptibly

public void lockInterruptibly() throws InterruptedException { 
        sync.acquireInterruptibly(1);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#acquireInterruptibly

public final void acquireInterruptibly(int arg)
            throws InterruptedException { 
        if (Thread.interrupted()) // 还没获取锁之前就中断了就直接抛出异常
            throw new InterruptedException();
        if (!tryAcquire(arg)) // 尝试获取锁，与不可中断的逻辑一致，区分公平与非公平
            doAcquireInterruptibly(arg);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#doAcquireInterruptibly

private void doAcquireInterruptibly(int arg)
        throws InterruptedException { 
        final Node node = addWaiter(Node.EXCLUSIVE); // 将当前线程封装成Node节点加入到同步队列尾部
        boolean failed = true;
        try { 
            for (;;) { 
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) { 
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException(); // 被中断了直接抛出异常
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

## 不可中断方法lock()对中断的处理 ##

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 演示独占锁中不可中断方法lock()对中断的处理 */
    public class InterruptThreadDemo2 { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            ReentrantLock reentrantLock = new ReentrantLock();
            new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t1").start();
    
            TimeUnit.SECONDS.sleep(1); // 等待t1启动
    
            Thread t2 = new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    System.out.println("t2: " + Thread.currentThread().isInterrupted()); // true
                }finally { 
                    reentrantLock.unlock();
                }
            }, "t2");
            t2.start();
    
            TimeUnit.SECONDS.sleep(1); // 等待t2启动
    
            t2.interrupt(); // 中断t2
            System.out.println("main: " + t2.isInterrupted()); // false
        }
    
    }

从运行结果可以发现，当对t2进行中断处理后，t2的中断标记位本应该为true，但是由于此时正在获取锁（不可中断），所以AQS将这个中断标记位先置为false，等待获取到锁后，重新将这个中断标记位还原为true。

下面从源码进行分析：

直接定位到java.util.concurrent.locks.AbstractQueuedSynchronizer\#acquireQueued

final boolean acquireQueued(final Node node, int arg) { 
        boolean failed = true;
        try { 
            boolean interrupted = false;
            for (;;) { 
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) { 
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true; // 这里只是记录了一下中断标记，如果是第一个等待的线程又会尝试获取一次锁，没获取到继续休眠，当这个线程获得锁后，会将中断标记状态返回
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

上面的方法执行完后返回到下面的方法。

java.util.concurrent.locks.AbstractQueuedSynchronizer\#acquire

public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt(); // 线程被中断了会进入到这里
    }
    
    static void selfInterrupt() { 
        Thread.currentThread().interrupt(); // 自己把自己中断，置中断标记位为true
    }

## 超时获取锁源码分析 ##

超时获取锁的使用如下：

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 演示独占锁的超时 */
    public class TimeoutThreadDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch countDownLatch = new CountDownLatch(1);
    
            ReentrantLock reentrantLock = new ReentrantLock();
            new Thread(()->{ 
                reentrantLock.lock();
                try { 
                    countDownLatch.countDown();
                    System.out.println("t1");
                    TimeUnit.SECONDS.sleep(30);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t1").start();
    
            countDownLatch.await();
    
            new Thread(()->{ 
                try { 
                    if(reentrantLock.tryLock(3, TimeUnit.SECONDS)) { 
                        System.out.println("t2 get lock");
                        reentrantLock.unlock();
                    } else { 
                        System.out.println("t2 get lock timeout");
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }, "t2").start();
        }
    
    }

java.util.concurrent.locks.ReentrantLock\#tryLock(long, java.util.concurrent.TimeUnit)

public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException { 
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#tryAcquireNanos

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException { 
        if (Thread.interrupted()) // 还没获取锁之前就中断了就直接抛出异常
            throw new InterruptedException();
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#doAcquireNanos

private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException { 
        if (nanosTimeout <= 0L)
            return false;
        final long deadline = System.nanoTime() + nanosTimeout; // 超时时间点
        final Node node = addWaiter(Node.EXCLUSIVE); // 将当前线程封装成Node节点加入到同步队列尾部
        boolean failed = true;
        try { 
            for (;;) { 
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) { 
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime(); // 剩余等待时间
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold) // 超时时间小于1000纳秒则直接自旋，不会休眠
                    LockSupport.parkNanos(this, nanosTimeout); // 带超时时间的休眠
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }