JDK源码——Lock&Condition类

摘要

涉及多线程问题，往往绕不开锁。在 JDK 1.5 之前，Java 通过 synchronized 关键字来实现锁的功能，该方式是语法层面的，由 JVM 实现。JDK 1.5 增加了锁在 API 层面的实现，也就是 java.util.concurrent.locks.Lock 接口及其相关的实现类，它不仅具备 synchronized 的功能，而且还增加了更加丰富的功能。通常与其配合使用的还有 Condition 接口。

Lock接口的定义

public interface Lock {
    // 阻塞式获取锁，该方法与synchronized功能类似
    void lock();

    // 获取锁，可响应中断
    void lockInterruptibly() throws InterruptedException;

    // 尝试获取锁，若成功返回true；否则返回false
    boolean tryLock();

    // 尝试获取锁（在给定的时间内），若成功返回true；否则返回false
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

    // 释放锁
    void unlock();

    // 创建一个与该锁绑定的Condition
    Condition newCondition();
}

Condition 接口定义如下：

public interface Condition {

    // 使当前线程等待，直到被signal唤醒或被中断
    void await() throws InterruptedException;

    // 使当前线程等待，直到被signal唤醒（不响应中断）
    void awaitUninterruptibly();

    // 使当前线程等待，直到被signal唤醒、或被中断、或到达等待时间
    long awaitNanos(long nanosTimeout) throws InterruptedException;

    // 使当前线程等待，直到被signal唤醒、或被中断、或到达等待时间（与上面方法类似）
    boolean await(long time, TimeUnit unit) throws InterruptedException;

    // // 使当前线程等待，直到被signal唤醒、或被中断、或到达给定的截止时间
    boolean awaitUntil(Date deadline) throws InterruptedException;

    // 唤醒一个等待的线程
    void signal();

    // 唤醒所有等待的线程
    void signalAll();
}

Condition 的方法虽然不少，但其实就两类：

• await* 方法：让当前线程处于等待状态；
• signal* 方法：唤醒处于等待的线程。

ReadWriteLock 接口：

定义了读锁和写锁，其中读锁是共享的，写锁是互斥的。

public interface ReadWriteLock {
    // 读锁
    Lock readLock();

    // 写锁
    Lock writeLock();
}

synchronized实战

public class ProdConsumerTest {
  private static final Object monitor = new Object();
  private Random random = new Random();
  private static final int SIZE = 10;
  private Queue<Integer> queue = new LinkedList<>();

  private void produce() throws InterruptedException {
    for (; ; ) {
       //使用 synchronized 锁
      synchronized (monitor) {
        if (queue.size() >= SIZE) {
          monitor.wait();
        }
        int nextInt = random.nextInt(1000);
        queue.offer(nextInt);

        sleep(400);
        System.out.println("size=" + queue.size() + ", 生产-->" + nextInt);
        //通知
        monitor.notify();
      }
    }
  }

  private void consume() throws InterruptedException {
    for (; ; ) {
      //使用 synchronized 锁
      synchronized (monitor) {
        if (queue.size() <= 0) {
          monitor.wait();
        }
        Integer poll = queue.poll();

        sleep(300);
        System.out.println("size=" + queue.size() + ", 消费成功-->" + poll);
        //通知        
        monitor.notify();
      }
    }
  }

  private void sleep(int timeout) {
    try {
      TimeUnit.MILLISECONDS.sleep(timeout);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public static void main(String[] args) {
    ProdConsumerTest test = new ProdConsumerTest();
    Thread t1 = new Thread(() -> {
      try {
        test.produce();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    Thread t2 = new Thread(() -> {
      try {
        test.consume();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    t1.start();
    t2.start();
  }
}

Lock/Condition 实现

public class ProdConsumerTest {
  private static final int SIZE = 10;
  private Random random = new Random();
  private Queue<Integer> queue = new LinkedList<>();

  // ReentrantLock 是 JDK 提供的 Lock 接口实现类，后文分析其原理
  private Lock lock = new ReentrantLock();
  private Condition notFull = lock.newCondition();
  private Condition notEmpty = lock.newCondition();

  private void produce() throws InterruptedException {
    for (; ; ) {
      //使用lock锁
      lock.lock();
      try {
        if (queue.size() >= SIZE) {
          notFull.await();
        }
        int nextInt = random.nextInt(1000);
        queue.offer(nextInt);

        sleep(400);
        System.out.println("size=" + queue.size() + ", 生产-->" + nextInt);
        notEmpty.signal();
      } finally {
        //使用unlock锁
        lock.unlock();
      }
    }
  }

  private void consume() throws InterruptedException {
    for (; ; ) {
      //使用lock锁
      lock.lock();
      try {
        if (queue.size() <= 0) {
          notEmpty.await();
        }
        Integer poll = queue.poll();

        sleep(300);
        System.out.println("size=" + queue.size() + ", 消费成功-->" + poll);
        notFull.signal();
      } finally {
        //使用unlock锁
        lock.unlock();
      }
    }
  }

  private void sleep(int timeout) {
    try {
      TimeUnit.MILLISECONDS.sleep(timeout);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public static void main(String[] args) {
    ProdConsumerTest test = new ProdConsumerTest();
    Thread t1 = new Thread(() -> {
      try {
        test.produce();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    Thread t2 = new Thread(() -> {
      try {
        test.consume();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    t1.start();
    t2.start();
  }
}

通过对比发现，Lock/Condition 可以实现和 synchronized 一样的功能，而 Condition 的 await/signal 则相当于 Object 的 wait/notify。

博文参考