Java源码之Condition

每天学Java

发布于 2020-06-02 09:54:57

8000

“ 在前面我们分析AQS和ReentrantLock的源码，在其中我们跳过了关于ConditionObject的分析，只说了它实现了Condition接口，那么今天我们就来看一下Condition这个接口是做什么的”

正式开始之前，我先回忆一下AQS这个构建锁的基础框架：通常来说我们需要重写tryAcquire方法去获取资源，tryRelease释放资源，举个例子：在ReentrantLock类中写了一个Sync内部类，这个内部类继承了抽象类AQS，重写了tryRelease方法，其中nonfairTryAcquire方法我们可以理解为tryAcquire方法，只不过由于ReentrantLock提供了公平锁和非公平锁，且非公平锁是默认的，公平锁中tryAcquire自己重写，非公平锁tryAcquire直接调用nonfairTryAcquire方法，所以Sync中就写了nonfairTryAcquire这个方法。可以参考：Java源码之AbstractQueuedSynchronizer Java源码之ReentrantLock 来看一下。

除此之外AQS中提供了newCondition方法返回一个Condition的实例。那么Condition这个接口到底提供了什么操作呢？我们现在来看一下：

—

首先：Condition接口中定义了7个方法：

这7个方法中：await()方法相当于Object的wait()方法，signal()方法相当于Object的notify()方法，signalAll()相当于Object的notifyAll()方法。但是Object中的wait(),notify(),notifyAll()方法是和"同步锁"(synchronized关键字)捆绑使用的；而Condition是需要与"互斥锁"/"共享锁"捆绑使用的。

// 造成当前线程在接到信号或被中断之前一直处于等待状态。
void await()
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
boolean await(long time, TimeUnit unit)
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
long awaitNanos(long nanosTimeout)
// 造成当前线程在接到信号之前一直处于等待状态。
void awaitUninterruptibly()
// 造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。
boolean awaitUntil(Date deadline)
// 唤醒一个等待线程。
void signal()
// 唤醒所有等待线程。
void signalAll()

Condition需要和互斥锁/共享锁绑定使用是什么意思呢？Condition需要和Lock配合使用，而不能单独拿出来用。除此之外：Condition能够更加精细的控制多线程的休眠与唤醒。对于同一个锁，我们可以创建多个Condition，在不同的情况下使用不同的Condition，怎么理解这句话呢，我们可以定义一个lock,但是可以得到多个Condition。

    private final Lock lock = new ReentrantLock();
    private final Condition pushCondition = lock.newCondition();
    private final Condition popCondition = lock.newCondition();

上面我们也说了他的方法和object中的wait(),notify(),notifyAll()方法类似，那我们就用Condition配合ReentrantLock实现生产者-消费者。

实现生产者和消费者的核心就是实现缓冲区，生产者向缓冲区中存放生产的对象，消费者则在缓冲区中取出对象。但是当缓冲区内存满了之后，生产者线程无法方法对象，应该阻塞，在被消费者消费之后，由消费者通知生产者生产。而当缓冲区没有对象后，消费者线程应该被阻塞，当生产者存放对象后，通知消费者消费。

那么我们先实现一个缓冲区：这个缓冲区用数组存放任务

public class Quenue<T> {
    //队列任务容量
    private int capacity;
    //用于计数。也可直接用obj.length
    private AtomicInteger currentNum = new AtomicInteger(0);
    //数组
    private Object[] obj = null;
    //定义锁
    private final Lock lock = new ReentrantLock();
    private final Condition pushCondition = lock.newCondition();
    private final Condition popCondition = lock.newCondition();

    //初始化队列容量
    public Quenue(int capacity) {
        this.capacity = capacity;
        init();
    }

    private void init() {
        obj = new Object[capacity];
    }


    public void putData(T t) {
        //上锁
        lock.lock();
        try {
            //队列满了
            while (currentNum.get() == capacity) {
                try {
                    System.out.println("队列满了，阻塞队列");
                    //生产者等待
                    pushCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //
            System.out.println("放入任务" + currentNum.get() + obj.length);
            obj[currentNum.get()] = t;
            currentNum.incrementAndGet();
            //唤醒消费者，注意这里是popCondition。
            popCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    public T take() {
        //上锁
        lock.lock();
        try {
            //队列为空
            while (currentNum.get() == 0) {
                try {
                    System.out.println("队列为空，等待");
                    //消费者等待
                    popCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            int x = currentNum.decrementAndGet();
            //唤醒生产者
            pushCondition.signal();
            return (T) obj[x];
        } finally {
            lock.unlock();
        }

    }

}

那么下面我们来制作生产者和消费者，为了方便观察，使用接口的方式来手动生产和消费任务。

@RestController
@RequestMapping("lock")
public class ConditionStudy {

    private Quenue<Runnable> quenue = new Quenue<>(3);

    @RequestMapping("producer/{id}")
    @ResponseBody
    public String pro(@PathVariable int id) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("方法ID=" + id + "的对象");
            }
        });
        quenue.putData(thread);
        return "放入成功,等待执行";
    }


    @RequestMapping("consumer")
    @ResponseBody
    public void cun() {
        while (true) {
            Thread o = (Thread) quenue.take();
            o.start();
        }
    }
}

我们启动项目首先不停的放入任务：我们可以看到由于我们缓冲区域容量为3，当我们放入第四个任务的时候，浏览器一直在加载，没有任何返回，说明被阻塞了。