JAVA并发（12）— Lock实现生产者消费者

无论是synchronize还是Lock，均可以将其看做为Monitor模型。

1. 条件变量&生产者消费者模型

在Monitor管程中，存在条件变量。

条件变量是管程内的一种数据结构，且只有在管程中才能访问它，它对管程内的所有过程是全局的。只能通过两个原子性操作来操纵它：

c.wait()：将调用线(进)程移入条件变量c所持有的队列中，并释放管程，直到另一个线(进)程在条件变量c上调用signal()方法释放线程。
c.signal()：会释放条件变量c队列上持有的阻塞线(进)程。

而在synchronized关键字中，在MonitorObject内置了一个条件变量。于是我们在同步块中会使用wait()或notify()方法后，会将同步块中的线程均放入到MonitorObject的Wait Set中，并进行阻塞。Wait Set中存在的是由于不同条件被阻塞的线程。会导致我们唤醒方法时，只能使用notifyAll()唤醒所有的线程。

public class SyncDemo {
    
    private List<String> list = new ArrayList<>(10);

    public void producer() {
        synchronized (this) {
            while (list.size() == 10) {
                try {
                    //若队列满了，将线程存入WaitSet中，等待被唤醒
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.add("产品");
            this.notifyAll();
        }
    }

    public void consumer() {
        synchronized (this) {
            while (list.size() == 0) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        list.remove(0);
        //唤醒生产者
        this.notifyAll();
    }
}

在Lock中，用户可以定义多个条件变量，即Condition，若未满足条件，则会存储到不同的条件变量所维护的队列中，以便可以唤醒不同条件队列的线程。

条件变量则允许线程由于一些未达到的条件而阻塞，此处的“条件”可以由用户来定义，在访问该条件时需要加锁（互斥量），如果条件没达到，线程将阻塞在该条件上。

public class Demo {
    //共享变量
    private List<String> list = new ArrayList<>(10);
    //互斥量
    private ReentrantLock lock = new ReentrantLock();
    //条件变量[不为空]。若不符合条件，则加入到条件变量到Condition queue中
    private Condition notEmptyCondition = lock.newCondition();
    //条件变量[不为满]。若不符合条件，则加入到条件变量到Condition queue中
    private Condition notFullCondition = lock.newCondition();
    /**
     * 生产者模型
     */
    public void producer() {
        try {
            lock.lock();
            //若list满了
            while (list.size() == 10) {
                try {
                    //线程进入，则将线程放入到不为full的条件变量中
                    //禁止再次生产。
                    notFullCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.add("元素");
            //唤醒生产者
            notEmptyCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消费者模型
     */
    public void consumer() {
        try {
            lock.lock();
            //若list为空
            while (list.size() == 0) {
                try {
                    //将该线程放入到条件变量中，该条件变量为不为空
                    //禁止再次消费
                    notEmptyCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.remove(0);
            notFullCondition.signal();
        } finally {
            lock.unlock();
        }
    }
}

2. ArrayBlockingQueue阻塞队列

阻塞队列ArrayBlockingQueue实际上也是使用生产者-消费者模型来实现的。

public class BlockingQueueDemo {
    private static BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(1);
    public static void main(String[] args) throws InterruptedException {
        /**
         * 增加元素
         */
        //如果队列满了，则抛出异常。
        blockingQueue.add("元素");
        //如果队列满了，则返回false。
        blockingQueue.offer("元素");
        //如果队列满了，则线程阻塞。
        blockingQueue.put("元素2");
        //如果队列满了，等待一段时间后依旧满，则返回false。
        blockingQueue.offer("元素2",1,TimeUnit.MICROSECONDS);
        /**
         * 移除元素
         */
        //如果队列为空，则线程阻塞。
        blockingQueue.take();
        //如果队列为空，则抛出异常。
        blockingQueue.remove();
        //如果队列为空，阻塞一段时间后依旧为null，则返回null。
        blockingQueue.poll(1, TimeUnit.MILLISECONDS);
        //如果队列为空，则返回null。
        blockingQueue.poll();
    }
}

生产者：当count为items.length数量时，即队列已满，不满足notFull的条件变量，所以线程会进入notFull所维护的条件队列中阻塞。

public void put(E e) throws InterruptedException {  
    checkNotNull(e);  
    final ReentrantLock lock = this.lock;  
    lock.lockInterruptibly();  
    try {  
        while (count == items.length)  
            notFull.await();  
        enqueue(e);  
    } finally {  
        lock.unlock();  
    }  
}

消费者：当items.length数量为0时。那么不满足notEmpty条件变量。该线程会加入到notEmpty所维护的条件队列中。

public E take() throws InterruptedException {  
    final ReentrantLock lock = this.lock;  
    lock.lockInterruptibly();  
    try {  
        while (count == 0)  
            notEmpty.await();  
        return dequeue();  
    } finally {  
        lock.unlock();  
    }  
}

JAVA并发（12）— Lock实现生产者消费者

JAVA并发（12）— Lock实现生产者消费者

1. 条件变量&生产者消费者模型

2. ArrayBlockingQueue阻塞队列

推荐阅读

相关阅读