JAVA并發(fā)（12）— Lock實(shí)現(xiàn)生產(chǎn)者消費(fèi)者

無論是synchronize還是Lock木人，均可以將其看做為Monitor模型。

1. 條件變量&生產(chǎn)者消費(fèi)者模型

在Monitor管程中，存在條件變量黔夭。

條件變量是管程內(nèi)的一種數(shù)據(jù)結(jié)構(gòu)，且只有在管程中才能訪問它捂贿，它對管程內(nèi)的所有過程是全局的纠修。只能通過兩個(gè)原子性操作來操縱它：

c.wait()：將調(diào)用線(進(jìn))程移入條件變量c所持有的隊(duì)列中，并釋放管程厂僧，直到另一個(gè)線(進(jìn))程在條件變量c上調(diào)用signal()方法釋放線程扣草。
c.signal()：會(huì)釋放條件變量c隊(duì)列上持有的阻塞線(進(jìn))程。

而在synchronized關(guān)鍵字中颜屠，在MonitorObject內(nèi)置了一個(gè)條件變量辰妙。于是我們在同步塊中會(huì)使用wait()或notify()方法后，會(huì)將同步塊中的線程均放入到MonitorObject的Wait Set中甫窟，并進(jìn)行阻塞密浑。Wait Set中存在的是由于不同條件被阻塞的線程。會(huì)導(dǎo)致我們喚醒方法時(shí)粗井，只能使用notifyAll()喚醒所有的線程尔破。

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

    public void producer() {
        synchronized (this) {
            while (list.size() == 10) {
                try {
                    //若隊(duì)列滿了，將線程存入WaitSet中浇衬，等待被喚醒
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.add("產(chǎn)品");
            this.notifyAll();
        }
    }

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

在Lock中懒构，用戶可以定義多個(gè)條件變量，即Condition耘擂，若未滿足條件胆剧，則會(huì)存儲到不同的條件變量所維護(hù)的隊(duì)列中，以便可以喚醒不同條件隊(duì)列的線程醉冤。

條件變量則允許線程由于一些未達(dá)到的條件而阻塞秩霍，此處的“條件”可以由用戶來定義篙悯，在訪問該條件時(shí)需要加鎖（互斥量），如果條件沒達(dá)到铃绒，線程將阻塞在該條件上鸽照。

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();
    /**
     * 生產(chǎn)者模型
     */
    public void producer() {
        try {
            lock.lock();
            //若list滿了
            while (list.size() == 10) {
                try {
                    //線程進(jìn)入椿疗，則將線程放入到不為full的條件變量中
                    //禁止再次生產(chǎn)漏峰。
                    notFullCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.add("元素");
            //喚醒生產(chǎn)者
            notEmptyCondition.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消費(fèi)者模型
     */
    public void consumer() {
        try {
            lock.lock();
            //若list為空
            while (list.size() == 0) {
                try {
                    //將該線程放入到條件變量中，該條件變量為不為空
                    //禁止再次消費(fèi)
                    notEmptyCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            list.remove(0);
            notFullCondition.signal();
        } finally {
            lock.unlock();
        }
    }
}

2. ArrayBlockingQueue阻塞隊(duì)列

阻塞隊(duì)列ArrayBlockingQueue實(shí)際上也是使用生產(chǎn)者-消費(fèi)者模型來實(shí)現(xiàn)的届榄。

public class BlockingQueueDemo {
    private static BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(1);
    public static void main(String[] args) throws InterruptedException {
        /**
         * 增加元素
         */
        //如果隊(duì)列滿了浅乔，則拋出異常。
        blockingQueue.add("元素");
        //如果隊(duì)列滿了铝条，則返回false靖苇。
        blockingQueue.offer("元素");
        //如果隊(duì)列滿了，則線程阻塞班缰。
        blockingQueue.put("元素2");
        //如果隊(duì)列滿了贤壁，等待一段時(shí)間后依舊滿，則返回false埠忘。
        blockingQueue.offer("元素2",1,TimeUnit.MICROSECONDS);
        /**
         * 移除元素
         */
        //如果隊(duì)列為空脾拆，則線程阻塞。
        blockingQueue.take();
        //如果隊(duì)列為空莹妒，則拋出異常名船。
        blockingQueue.remove();
        //如果隊(duì)列為空，阻塞一段時(shí)間后依舊為null旨怠，則返回null渠驼。
        blockingQueue.poll(1, TimeUnit.MILLISECONDS);
        //如果隊(duì)列為空，則返回null鉴腻。
        blockingQueue.poll();
    }
}

生產(chǎn)者：當(dāng)count為items.length數(shù)量時(shí)迷扇，即隊(duì)列已滿，不滿足notFull的條件變量爽哎，所以線程會(huì)進(jìn)入notFull所維護(hù)的條件隊(duì)列中阻塞谋梭。

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();  
    }  
}

消費(fèi)者：當(dāng)items.length數(shù)量為0時(shí)。那么不滿足notEmpty條件變量倦青。該線程會(huì)加入到notEmpty所維護(hù)的條件隊(duì)列中。

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

JAVA并發(fā)（12）— Lock實(shí)現(xiàn)生產(chǎn)者消費(fèi)者

JAVA并發(fā)（12）— Lock實(shí)現(xiàn)生產(chǎn)者消費(fèi)者

1. 條件變量&生產(chǎn)者消費(fèi)者模型

2. ArrayBlockingQueue阻塞隊(duì)列

推薦閱讀

相關(guān)閱讀