Java并发-locks包源码剖析1-Lock和Reentran

作者: 宛丘之上兮 | 来源:发表于2018-11-29 21:38 被阅读5次

前面几篇文章分析了java.util.concurrent.atomic包下的原子类和synchronized同步锁，这篇分析JUC的locks包下的锁类。java.util.concurrent.locks下的类不是很多，但是比较复杂，定义了基本的锁Lock，对线程进行park和unPark的LockSupport和核心的AQS框架（AbstractQueuedSynchronizer）。

$\color{blue}{1\ Lock}$

先看下Lock的源码：

public interface Lock {
    void lock();
    void lockInterruptibly() throws InterruptedException;
    boolean tryLock();
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    void unlock();
    Condition newCondition();
}

是只有六个方法的接口。Lock和synchronized具有同样的含义和功能，synchronized 锁在退出块时自动释放，而Lock 需要手动释放，Lock更加灵活。synchronized 是系统关键字，Lock则是jdk1.5以来提供的一个接口。

synchronized缺点很明显：一个正在等候获得synchronized锁的线程无法被中断；也无法通过投票得到锁，如果想要得到锁那么就必须得等下去直到释放锁；synchronized还要求锁的释放只能在与获得锁所在的堆栈帧相同的堆栈帧中进行。
而Lock（如ReentrantLock ）除了与Synchronized 具有相同的语义外，还支持锁投票、定时锁等候和可中断锁等候(就是说在等待锁的过程中，可以被中断)的一些特性。调用lockInterruptibly后，或者获得锁，或者被中断后抛出异常。优先响应异常。

Lock 接口有 3 个实现它的类：ReentrantLock、ReetrantReadWriteLock.ReadLock 和 ReetrantReadWriteLock.WriteLock，即重入锁、读锁和写锁，下面介绍下ReentrantLock。

$\color{blue}{2\ ReentrantLock}$

ReentrantLock，意思是“可重入锁”，ReentrantLock实现了Lock接口，并且ReentrantLock提供了更多的方法。ReentrantLock的锁内部实现通过NonfairSync和FairSync实现，它提供了两个构造器：

    public ReentrantLock() {
        sync = new NonfairSync();
    }

    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

无参数构造器采用默认的NonfairSync机制，第二个构造器根据参数来决定使用公平锁还是非公平锁。

synchronized 采用的同步策略称为阻塞同步，它属于一种悲观的并发策略，即线程获得的是独占锁。独占锁意味着其他线程只能依靠阻塞来等待线程释放锁。而在 CPU 转换线程阻塞时会引起线程上下文切换，当有很多线程竞争锁的时候，会引起 CPU 频繁的上下文切换导致效率很低。

随着指令集的发展，我们有了另一种选择：基于冲突检测的乐观并发策略，通俗地讲就是先进性操作，如果没有其他线程争用共享数据，那操作就成功了，如果共享数据被争用，产生了冲突，那就再进行其他的补偿措施（最常见的补偿措施就是不断地自旋重拾，直到试成功为止），这种乐观的并发策略的许多实现都不需要把线程挂起，因此这种同步被称为非阻塞同步。ReetrantLock 采用的便是这种并发策略加上LockSupport提供的park/unPark操作。

“欲知天道，察其数”，想知其原理，先看其表现，那么我们先通过例子看下如何正确使用ReentrantLock，然后再探究其原理。测试代码如下：

public class Fs {
    private int cnt = 0;

    public static void main(String args[]) {
        Fs fs = new Fs();
        int threadCnt = 10;
        Thread[] threads = new Thread[threadCnt];
        CountDownLatch cdt = new CountDownLatch(threadCnt);
        for (int i = 0; i < threadCnt; i++) {
            threads[i] = new Thread(() -> {
                Lock lock = new ReentrantLock();
                lock.lock();
                try {
                    for (int j = 0; j < 10000; j++) {
                        fs.add();
                    }
                } catch (Exception e) {
                } finally {
                    lock.unlock();
                }
                cdt.countDown();
            });
            threads[i].start();
        }
        while (Thread.activeCount() > 1) {
            Thread.yield();
        }
//        for (Thread i : threads) {
//            try {
//                i.join();
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
//        try {
//            cdt.await();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        System.out.println(fs.cnt + "  ");
    }

    private void add() {
        ++cnt;
    }
}

Java并发-synchronized从入门到精通这篇文章结尾介绍了JVM对锁优化的手段包括锁清除，根据锁清除和代码逃逸原理，各位朋友猜猜看上面的代码能输出我们期望的值10,000吗？

答案是不能。lock属于一个局部变量不会从当前线程中逃逸出去，因此也不会被其他线程所使用，因此不可能存在共享资源竞争的情景，JVM会自动将其锁消除。所以输出结果总是小于10,000。正确的代码如下：

public class Fs {
    private int cnt = 0;
    private Lock lock = new ReentrantLock();

    public static void main(String args[]) {
        Fs fs = new Fs();
        int threadCnt = 10;
        Thread[] threads = new Thread[threadCnt];
        CountDownLatch cdt = new CountDownLatch(threadCnt);
        for (int i = 0; i < threadCnt; i++) {
            threads[i] = new Thread(() -> {
                fs.lock.lock();
                try {
                    for (int j = 0; j < 10000; j++) {
                        fs.add();
                    }
                } catch (Exception e) {
                } finally {
                    fs.lock.unlock();
                }
                cdt.countDown();
            });
            threads[i].start();
        }
        while (Thread.activeCount() > 1) {
            Thread.yield();
        }
        System.out.println(fs.cnt + "  ");
    }

    private void add() {
        ++cnt;
    }
}

关于lock()方法的使用，需要注意以下几点：

lock()和unlock()必须成对出现，如果只出现lock()会导锁不会被释放，如果只出现unlock()会抛出异常java.lang.IllegalMonitorStateException at java.util.concurrent.locks.ReentrantLock$Sync.tryRelease(ReentrantLock.java:151) at java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1261) at java.util.concurrent.locks.ReentrantLock.unlock(ReentrantLock.java:457) at Fs.lambda$main$0(Fs.java:25) at java.lang.Thread.run(Thread.java:745)。
lock()和unlock()之间的同步代码块一定要用try-catch处理，就算你百分之一万的确定同步代码块不会发生异常，也最好加上try-catch包裹起来，如果不用try-catch，那么万一出现异常的话，unlock就不会执行而导致锁无法被释放，而且unlock要放到finally语句里。
lock()不会响应中断，如果想要响应中断，需要使用lockInterruptibly()方法。

关于synchronized和Lock，需要再提一下：

等待synchronized锁的线程无法被中断(中断标记位无法被设置为true)，获得synchronized锁的线程可以被中断；
等待Lock锁的线程可以被中断(这里是指中断标记位被设置为true)，但是lock()方法无法响应中断，lockInterruptibly()可以响应中断。

关于线程的中断，你需要理解这三个方法才能继续往下阅读：

//中断线程（实例方法）
public void Thread.interrupt();
//判断线程是否被中断（实例方法）
public boolean Thread.isInterrupted();
//判断是否被中断并清除当前中断状态（静态方法）
public static boolean Thread.interrupted();

为了证明等待Lock锁的线程可以被中断，我写了个程序专门测试下：

public class Fg implements Runnable {
    @Override
    public void run() {
        lock.lock();
        f();
        //中断判断
        System.out.println("等待锁线程执行完毕" + Thread.currentThread().isInterrupted());
        lock.unlock();
    }

    Fg() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                lock.lock();
                f();
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("获得了锁的线程执行完毕");
                lock.unlock();
            }
        }, "获得了锁的线程开始执行").start();
    }

    private void f() {
        System.out.println(Thread.currentThread().getName() + " " + Math.random());
    }

    private int cnt = 0;
    private Lock lock = new ReentrantLock();

    public static void main(String args[]) {
        Fg fg = new Fg();
        Thread t = new Thread(fg, "等待锁线程开始执行");
        t.start();
        t.interrupt();
        System.out.println(t.isInterrupted());
    }
}

代码输出是：

true
获得了锁的线程开始执行 0.40285651579626425
获得了锁的线程执行完毕
等待锁线程开始执行 0.8003535914600582
等待锁线程执行完毕true

上述代码中，线程t启动前，它的内部已经启动了一个内部线程并且获得了lock锁，t启动的时候也企图获得lock锁，因此会被阻塞，然后调用t.interrupt();中断t，日志显示中断标记为已经是true说明中断成功，内部线程3秒后释放锁，线程t获得锁后执行后面的代码。可以看到，等待锁的线程t能被中断，但是如果调用lock()方法获取锁失败，也是会自旋重试的(外观效果和阻塞是一样的，但是CAS的自旋比阻塞高效很多)，直到获取锁成功，如果想要响应中断而不被自旋等待，需要使用lockInterruptibly方法。

为了证明获得Lock锁的线程也能被中断(lock()方法同样无法响应中断，lockInterruptibly()可以响应中断，我这里的被中断意思是说中断标记位被成功设置为true)，我又写了个程序专门测试下：

public class Fs {
    private int cnt = 0;
    private Lock lock = new ReentrantLock();

    public static void main(String args[]) {
        Fs fs = new Fs();
        int threadCnt = 1;
        Thread[] threads = new Thread[threadCnt];
        CountDownLatch cdt = new CountDownLatch(threadCnt);
        for (int i = 0; i < threadCnt; i++) {
            threads[i] = new Thread(() -> {
                fs.lock.lock();
                try {
                    while(true) {
                        if (Thread.currentThread().isInterrupted()) {//这个if不会重置中断标记位，想要重置中断标记位需要使用Thread.interrupted()
                            System.out.println("中断线程!! " + Thread.currentThread().isInterrupted() + " " + threads[0].isInterrupted());
                            break;
                        } else {
                            System.out.println(Thread.currentThread().getName() + Math.random());
                        }
                    }
                } catch (Exception e) {
                    System.out.println("异常了：" + e.toString());
                } finally {
                    fs.lock.unlock();
                }
                cdt.countDown();
            });
            threads[i].start();
        }
        threads[0].interrupt();
        System.out.println("threads[0].isInterrupted():   " + threads[0].isInterrupted());
        System.out.println("fs.cnt   " + fs.cnt);
    }
}

输出结果是(输出顺序不固定，因为三个输出的代码是运行在两个线程中的)

threads[0].isInterrupted():   true
中断线程!! true true
fs.cnt   0

输出结果给我们两点提示

threads[0]获得了锁，而且中断标记位能被设置为true；
可以看到被中断的线程threads[0]的中断标记位一直是true，因此我们知道interrupt不会重置中断标记位，如果想要重置中断标记位，那么需要if条件需要使用Thread.interrupted()而不是Thread.currentThread().isInterrupted()，在while(true)后面的if条件语句你可以使用Thread.interrupted()代替，看下输出结果回是这样的：

threads[0].isInterrupted():   true
中断线程!! false false
fs.cnt   0

即中断标记位被重置(为啥第一行输出的标记位是true，我猜测是Thread.interrupted()重置标记位需要一定的时间，即还没有重置为false第一行日志就打印出来了，如果在最后延迟1秒后再打印threads[0]的中断标记位，就是false了，这个猜测我是亲自试过的，是对的)。

tryLock()方法是有返回值的，它表示用来尝试获取锁，如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false，也就说这个方法无论如何都会立即返回。在拿不到锁时不会一直在那等待。
tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的，只不过区别在于这个方法在拿不到锁时会等待一定的时间，在时间期限之内如果还拿不到锁，就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁，则返回true。

$\color{blue}{3\ ReentrantLock}$ 条件变量实现线程间协作

synchronized可以配合使用 Object 对象的 wait（）和 notify()或 notifyAll()方法来实现线程间协作。Java 5 之后，我们可以用 Reentrantlock 锁配合 Condition 对象上的 await()和 signal()或 signalAll()方法来实现线程间协作。在 ReentrantLock 对象上 newCondition()可以得到一个 Condition 对象，可以通过在 Condition 上调用 await()方法来挂起一个任务（线程），通过在 Condition 上调用 signal()来通知任务，从而唤醒一个任务，或者调用 signalAll()来唤醒所有在这个 Condition 上被其自身挂起的任务。另外，如果使用了公平锁，signalAll()的与 Condition 关联的所有任务将以 FIFO 队列的形式获取锁，如果没有使用公平锁，则获取锁的任务是随机的，这样我们便可以更好地控制处在 await 状态的任务获取锁的顺序。与 notifyAll()相比，signalAll()是更安全的方式。另外，它可以指定唤醒与自身 Condition 对象绑定在一起的任务。

下面是生产者——消费者模型的代码：

class Info { // 定义信息类
    private String name = "name";//定义name属性，为了与下面set的name属性区别开
    private String content = "content";// 定义content属性，为了与下面set的content属性区别开
    private boolean flag = true;   // 设置标志位,初始时先生产
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition(); //产生一个Condition对象

    public void set(String name, String content) {
        lock.lock();
        try {
            while (!flag) {
                condition.await();
            }
            this.name = name;    // 设置名称
            this.content = content;  // 设置内容
            Thread.sleep(300);
            System.out.println("生产 " + this.name + " --> " + this.content);
            flag = false; // 改变标志位，表示可以取走
            condition.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void get() {
        lock.lock();
        try {
            while (flag) {
                condition.await();
            }
            System.out.println("消费 " + this.name + " --> " + this.content);
            Thread.sleep(300);
            flag = true;  // 改变标志位，表示可以生产
            condition.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

class Producer implements Runnable { // 通过Runnable实现多线程
    private Info info = null;      // 保存Info引用

    public Producer(Info info) {
        this.info = info;
    }

    public void run() {
        for (int i = 0; i < 10; i++) {
            this.info.set("姓名--" + i, "内容--" + +i);    // 设置名称
        }
    }
}

class Consumer implements Runnable {
    private Info info = null;

    public Consumer(Info info) {
        this.info = info;
    }

    public void run() {
        for (int i = 0; i < 10; i++) {
            this.info.get();
        }
    }
}

public class ThreadCaseDemo {
    public static void main(String args[]) {
        Info info = new Info(); // 实例化Info对象
        Producer pro = new Producer(info); // 生产者
        Consumer con = new Consumer(info); // 消费者
        new Thread(pro).start();
        //启动了生产者线程后，再启动消费者线程
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(con).start();
    }
}

输出：

生产 姓名--0 --> 内容--0
消费 姓名--0 --> 内容--0
生产 姓名--1 --> 内容--1
消费 姓名--1 --> 内容--1
生产 姓名--2 --> 内容--2
消费 姓名--2 --> 内容--2
生产 姓名--3 --> 内容--3
消费 姓名--3 --> 内容--3
生产 姓名--4 --> 内容--4
消费 姓名--4 --> 内容--4
生产 姓名--5 --> 内容--5
消费 姓名--5 --> 内容--5
生产 姓名--6 --> 内容--6
消费 姓名--6 --> 内容--6
生产 姓名--7 --> 内容--7
消费 姓名--7 --> 内容--7
生产 姓名--8 --> 内容--8
消费 姓名--8 --> 内容--8
生产 姓名--9 --> 内容--9
消费 姓名--9 --> 内容--9

上面代码通过条件变量condition的await()和signal()达到生产者线程和消费者线程间的同步与协作。

$\color{blue}{4\ ReentrantLock}$ 与 synchronized 性能比较

在 JDK1.5 中，synchronized 是性能低效的。因为这是一个重量级操作，它对性能最大的影响是阻塞的是实现，挂起线程和恢复线程的操作都需要转入内核态中完成，这些操作给系统的并发性带来了很大的压力。相比之下使用Java 提供的 Lock 对象，性能更高一些。Brian Goetz 对这两种锁在 JDK1.5、单核处理器及双 Xeon 处理器环境下做了一组吞吐量对比的实验，发现多线程环境下，synchronized的吞吐量下降的非常严重，而ReentrankLock 则能基本保持在同一个比较稳定的水平上。但与其说 ReetrantLock 性能好，倒不如说 synchronized 还有非常大的优化余地，于是到了 JDK1.6，发生了变化，对 synchronize 加入了很多优化措施，有自适应自旋，锁消除，锁粗化，轻量级锁，偏向锁等等。导致在 JDK1.6 上 synchronize 的性能并不比 Lock 差。官方也表示，他们也更支持 synchronize，在未来的版本中还有优化余地，所以还是提倡在 synchronized 能实现需求的情况下，优先考虑使用 synchronized 来进行同步。

前面讲了很多，但是并没有解释ReentrantLock的锁机制，关于ReentrantLock的锁机制，我打算放到下一篇文章中分析，不然文章太长，读起来会不会觉得有点累。

参考文献

并发新特性—Lock 锁与条件变量

Java并发-locks包源码剖析1-Lock和Reentran

$\color{blue}{1\ Lock}$

$\color{blue}{2\ ReentrantLock}$

$\color{blue}{3\ ReentrantLock}$ 条件变量实现线程间协作

$\color{blue}{4\ ReentrantLock}$ 与 synchronized 性能比较

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

程序员