Java并发——Lock锁(七)

原创

翰墨飘香

修改于 2024-04-21 07:55:19

1310

修改于 2024-04-21 07:55:19

文章被收录于专栏：Java并发Java并发

一、Lock简介

在Java中，Lock是一个接口，它提供了比synchronized关键字更灵活的线程同步机制。Lock接口的常用实现类是ReentrantLock和ReadWriteLock

二、Lock基本方法

方法名称	方法描述
void lock()	获得锁。如果锁已经被其他线程获取，则会被阻塞直至锁释放
void lockInterruptibly()	获取锁并允许被中断，和 tryLock(long time, TimeUnit unit) 方法不同的是它允许被中断并抛出中断异常。
boolean tryLock()	尝试获取锁。会立即返回结果，而不会被阻塞。如果可用，则获取锁定，并立即返回值为true；如果锁不可用，则此方法将立即返回值为false 。
boolean tryLock(long time, TimeUnit unit)	尝试获取锁并等待一段时间，当前线程在一下三种情况下会返回： 1. 当前线程在超时时间内获得了锁；2.当前线程在超时时间内被中断；3.超时时间结束，返回false.
Condition newCondition()	获取等待通知组件，该组件和当前的锁绑定，当前线程只有获得了锁，才能调用该组件的wait()方法，而调用后，当前线程将释放锁。
void unlock()	释放锁

三、 ReentrantLock

3.1 ReentrantLock原理

底层使用CAS+AQS队列来实现

参考：

Java并发--AQS

ReentrantLock原理

https://juejin.cn/post/7023050383417147422?searchId=20240407174524F046BFE9E681E213711C

ReentrantLock源码解析

1、构造方法

ReentranLock有两个构造方法

   /**
     *  默认构造,非公平锁
     */
    public ReentrantLock() {
        sync = new NonfairSync();
    }

    /**
     * 通过boolean值控制 FairSync时公平锁
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

2、加锁流程源码

NonfairSync 和FairSync 都继承自 Sync ，Sync又继承自AQS

①下面以非公平锁NonfairSync为例子解析下源码，上来先尝试将state从0修改为1，如果成功，代表获取锁资源。如果没有成功，调用acquire。state是AQS中的一个由volatile修饰的int类型变量，多个线程会通过CAS的方式修改state，在并发情况下，只会有一个线程成功的修改state。

   static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

② acquire()方法：CAS加锁失败进入acquire()方法，acquire是一个业务方法，里面并没有实际的业务处理，都是在调用其他方法

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

③tryAcquire方法

protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
   //拿到当前线程
            final Thread current = Thread.currentThread();
            //拿到AQS的state
            int c = getState();
            // 如果state == 0，说明没有线程占用着当前的锁资源
            if (c == 0) {
            //获取锁资源
                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;
        }

④acquireQueued&addWaiter方法：在获取锁资源失败后，需要将当前线程封装为Node对象，并且插入到AQS队列的末尾。

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);
        }
    }
private Node addWaiter(Node mode) {
    // 将当前线程封装为Node对象，mode为null，代表互斥锁
    Node node = new Node(Thread.currentThread(), mode);
    // pred是tail节点
    Node pred = tail;
    // 如果pred不为null，有线程正在排队
    if (pred != null) {
        // 将当前节点的prev，指定tail尾节点
        node.prev = pred;
        // 以CAS的方式，将当前节点变为tail节点
        if (compareAndSetTail(pred, node)) {
            // 之前的tail的next指向当前节点
            pred.next = node;
            return node;
        }
    }
    // 添加的流程为，  自己prev指向、tail指向自己、前节点next指向我
    // 如果上述方式，CAS操作失败，导致加入到AQS末尾失败，如果失败，就基于enq的方式添加到AQS队列
    enq(node);
    return node;
}
// enq，无论怎样都添加进入
private Node enq(final Node node) {
    for (;;) {
        // 拿到tail
        Node t = tail;
        // 如果tail为null，说明当前没有Node在队列中
        if (t == null) { 
            // 创建一个新的Node作为head，并且将tail和head指向一个Node
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 和上述代码一致！
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

3、解锁流程源码

释放锁资源，将state减1,如果state减为0了，唤醒在队列中排队的Node

public final boolean release(int arg) {
    // 核心的释放锁资源方法
    if (tryRelease(arg)) {
        // 释放锁资源释放干净了。  （state == 0）
        Node h = head;
        // 如果头节点不为null，并且头节点的状态不为0，唤醒排队的线程
        if (h != null && h.waitStatus != 0)
            // 唤醒线程
            unparkSuccessor(h);
        return true;
    }
    // 释放锁成功，但是state != 0
    return false;
}
// 核心的释放锁资源方法
protected final boolean tryRelease(int releases) {
    // 获取state - 1
    int c = getState() - releases;
    // 如果释放锁的线程不是占用锁的线程，抛异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    // 是否成功的将锁资源释放利索 （state == 0）
    boolean free = false;
    if (c == 0) {
        // 锁资源释放干净。
        free = true;
        // 将占用锁资源的属性设置为null
        setExclusiveOwnerThread(null);
    }
    // 将state赋值
    setState(c);
    // 返回true，代表释放干净了
    return free;
}


// 唤醒节点
private void unparkSuccessor(Node node) {
    // 拿到头节点状态
    int ws = node.waitStatus;
    // 如果头节点状态小于0，换为0
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    // 拿到当前节点的next
    Node s = node.next;
    // 如果s == null ，或者s的状态为1
    if (s == null || s.waitStatus > 0) {
        // next节点不需要唤醒，需要唤醒next的next
        s = null;
        // 从尾部往前找，找到状态正常的节点。（小于等于0代表正常状态）
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    // 经过循环的获取，如果拿到状态正常的节点，并且不为null
    if (s != null)
        // 唤醒线程
        LockSupport.unpark(s.thread);
}

4、使用实例

lock方法

public class ReentrantLockDemo {
    private Lock lock = new ReentrantLock();

    public void test() {
        lock.lock();
        try {
            System.out.println("ThreadName=" + Thread.currentThread().getName());
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }
}

class Main {
    public static void main(String[] args) {
        ReentrantLockDemo lockTest = new ReentrantLockDemo();
        for (int i = 0; i < 5; i++) {
            new Thread(lockTest::test, "Thread-" + i).start();
        }
    } 
}

tryLock

public class ReentrantLockTest {
    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        new Thread(()->test(lock),"线程A").start();
        new Thread(()->test(lock),"线程B").start();
    }
    public static   void test(ReentrantLock lock){
            try {
                if(lock.tryLock(2, TimeUnit.SECONDS)){
                    try {
                        System.out.println(Thread.currentThread().getName()+"获取了锁！");
                        Thread.sleep(3000);
                    }finally {
                        lock.unlock();
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }


    }

3.2 ReadWriteLock

1、ReadWriteLock概述

分为读锁（Read Lock）和写锁（Write Lock）。读锁是共享的，多个线程可以同时持有读锁。而写锁则是独占的，一旦一个线程获取了写锁，其他线程就只能等它写完。

深度解析Java中的ReadWriteLock：高效处理并发读写操作

2、读写锁的获取规则

①如果有一个线程已经占用了读锁，则此时其他线程如果要申请读锁，可以申请成功。

②如果有一个线程已经占用了读锁，则此时其他线程如果要申请写锁，则申请写锁的线程会一直等待释放读锁，因为读写不能同时操作。

③如果有一个线程已经占用了写锁，则此时其他线程如果申请写锁或者读锁，都必须等待之前的线程释放写锁，同样也因为读写不能同时，并且两个线程不应该同时写。

一句话总结：要么是一个或多个线程同时有读锁，要么是一个线程有写锁，但是两者不会同时出现。也可以总结为：读读共享、其他都互斥（写写互斥、读写互斥、写读互斥）。

2、ReentrantReadWriteLock原理

① 读写两把锁

ReentrantReadWriteLock 里面有两把锁，一个读锁，一个写锁，都继承自Sync类

② 读写锁共享state变量

是基于AQS来实现的，但是我们发现AQS里面只有一个state变量，可以进行维护一把锁，那么你读写锁有两把锁是如何基于我AQS来实现的呢？答案是按位切分使用，一个state变量是int型，32位。然后我高16位读锁使用，低16位写锁使用

    private static final long serialVersionUID = 6317671515068378041L;

        /*
         * Read vs write count extraction constants and functions.
         * Lock state is logically divided into two unsigned shorts:
         * The lower one representing the exclusive (writer) lock hold count,
         * and the upper the shared (reader) hold count.
         */

        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

        /** Returns the number of shared holds represented in count  */
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        /** Returns the number of exclusive holds represented in count  */
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

当我们使用读锁，假如当前同步状态位w，那么在高16位上+1（读锁），w+(1<<16)；

我们使用写锁，获取读锁状态，w>>>16，能够获取到低16位。

3、ReadWriteLock适合场景

适合场景

1、读多写少

当一个应用主要涉及到读取操作，而写操作相对较少时，使用ReadWriteLock非常合适。因为它允许多个线程同时读取数据，从而大大提高了并发性能。这就像图书馆里的一本热门书籍，大家都在阅读，但只有偶尔有人在做笔记

2、数据一致性要求高

在需要确保数据在读取时不被修改的场景中，ReadWriteLock也很适用。它通过写锁来保证在写操作进行时，读操作必须等待，从而保证了数据的一致性

不适合场景：

1、写操作频繁

如果一个应用中写操作非常频繁，使用ReadWriteLock可能就不是最佳选择了。因为频繁的写操作会导致读操作频繁地等待，从而降低程序的总体性能。

2、资源竞争不激烈

在线程间的资源竞争不是很激烈的场景中，使用简单的互斥锁（例如ReentrantLock）可能就足够了。在这种情况下，ReadWriteLock的复杂性可能并不会带来额外的好处。

4、读写锁例子

public class ReentrantReadWriteLockDemo {
    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> read()).start();
        new Thread(() -> read()).start();
        new Thread(() -> write()).start();
        new Thread(() -> write()).start();
    }
    private static final ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock(false);
    private static final ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
    private static final ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();

    private static void read() {
        readLock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + "得到读锁，正在读取");
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(Thread.currentThread().getName() + "释放读锁");
            readLock.unlock();
        }
    }

    private static void write() {
        writeLock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + "得到写锁，正在写入");
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();} finally {
            System.out.println(Thread.currentThread().getName() + "释放写锁");
            writeLock.unlock();
        }
    }

}

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

java