0
点赞
收藏
分享

微信扫一扫

JAVA并发(12)— Lock实现生产者消费者

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

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

在Monitor管程中,存在条件变量

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

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

而在synchronized关键字中,在MonitorObject内置了一个条件变量。于是我们在同步块中会使用wait()notify()方法后,会将同步块中的线程均放入到MonitorObjectWait 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并发(1)—java对象布局
JAVA并发(2)—PV机制与monitor(管程)机制
JAVA并发(3)—线程运行时发生GC,会回收ThreadLocal弱引用的key吗?
JAVA并发(4)— ThreadLocal源码角度分析是否真正能造成内存溢出!
JAVA并发(5)— 多线程顺序的打印出A,B,C(线程间的协作)
JAVA并发(6)— AQS源码解析(独占锁-加锁过程)
JAVA并发(7)—AQS源码解析(独占锁-解锁过程)
JAVA并发(8)—AQS公平锁为什么会比非公平锁效率低(源码分析)
JAVA并发(9)— 共享锁的获取与释放
JAVA并发(10)—interrupt唤醒挂起线程
JAVA并发(11)—AQS源码Condition阻塞和唤醒
JAVA并发(12)— Lock实现生产者消费者

举报

相关推荐

0 条评论