Java併發編程初級篇(十六)：Lock+Condition實現生產者消費者問題

以前咱們在「Java併發編程初級篇(十二)：使用wait和notify生產者消費者問題」，已經使用Java提供的synchronized關鍵字和wait()，notify()，notifyAll()方法實現過來生產者消費者問題。Java API還爲咱們提供了鎖的解決方案。

使用鎖解決阻塞要用到Condition，它是經過Lock.newCondition()來得到的。就像wait()和notify()必須在synchronized塊內同樣，Condition.await()和Condition.singialAll()方法也必須在Lock.lock()和Lock.unlock()內執行。

代碼示例：

首先咱們實現一個數據緩衝區，緩衝區中定義了maxSize變量來表明緩衝區大小，LinkedList來模擬緩衝區。而後添加一把鎖，並用這把鎖來新建兩個Condition：producer(控制生產者掛起和喚醒)和consumer(控制消費者掛起和喚醒)。當生產者發現緩衝區滿的狀況下調用producer.await()掛起，等待消費者消費數據後調用producer.singialAll()方法來喚醒，並從新判斷緩衝區狀態。當消費者發現緩衝區空的狀況下調用consumer.await()掛起，等待生產者向緩衝區中放入數據並調用consumer.singialAll()方法喚醒並從新判斷緩衝區狀態。

public class DataBuffer {
    private int maxSize;
    private LinkedList<Date> buffer;

    private Lock lock;
    private Condition producer;
    private Condition consumer;

    public DataBuffer(int maxSize) {
        this.maxSize = maxSize;
        buffer = new LinkedList<Date>();

        lock = new ReentrantLock();

        producer = lock.newCondition();
        consumer = lock.newCondition();
    }

    public void put() {
        try {
            lock.lock();

            while (buffer.size() == this.maxSize) {
                producer.await();
            }

            buffer.add(new Date());

            System.out.printf("%s: Add one %s. Buffer size is %d.\n",
                    Thread.currentThread().getName(),
                    buffer.peekLast(),
                    buffer.size());

            consumer.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void get() {
        try {
            lock.lock();

            while (buffer.size() == 0) {
                consumer.await();
            }

            System.out.printf("%s: Get one %s. Buffer size is %d.\n",
                    Thread.currentThread().getName(),
                    buffer.pollFirst(),
                    buffer.size());

            producer.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

建立兩個線程用於模擬生產者和消費者

public class Producer implements Runnable {
    private DataBuffer buffer;

    public Producer(DataBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        buffer.put();
    }
}

public class Consumer implements Runnable {
    private DataBuffer buffer;

    public Consumer(DataBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        buffer.get();
    }
}

建立兩個任務類線程啓動10個生產者和消費者，生產者生產速度>消費者消費速度。

public class ProducerTask implements Runnable {
    private DataBuffer buffer;

    public ProducerTask(DataBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            new Thread(new Producer(buffer), "Producer-" + i).start();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class ConsumerTask implements Runnable {
    private DataBuffer buffer;

    public ConsumerTask(DataBuffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            new Thread(new Consumer(buffer), "Consumer-" + i).start();
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

主方法類中啓動兩個任務類線程，模擬生產者和消費者，由於生產者速度快，最終會致使生產者處於阻塞狀態。

public class Main {
    public static void main(String[] args) {
        DataBuffer buffer = new DataBuffer(5);

        new Thread(new ProducerTask(buffer), "ProducerTask").start();
        new Thread(new ConsumerTask(buffer), "ConsumerTask").start();
    }
}

日誌，由於緩衝區大小爲5，因此當緩衝區內容達到5個以後，生產者進入阻塞狀態，並等待消費者消費數據後被喚醒並向緩衝區中插入數據。

Producer-0: Add one Fri Nov 25 15:36:21 CST 2016. Buffer size is 1.
Consumer-0: Get one Fri Nov 25 15:36:21 CST 2016. Buffer size is 0.
Producer-1: Add one Fri Nov 25 15:36:22 CST 2016. Buffer size is 1.
Producer-2: Add one Fri Nov 25 15:36:23 CST 2016. Buffer size is 2.
Producer-3: Add one Fri Nov 25 15:36:24 CST 2016. Buffer size is 3.
Producer-4: Add one Fri Nov 25 15:36:25 CST 2016. Buffer size is 4.
Consumer-1: Get one Fri Nov 25 15:36:22 CST 2016. Buffer size is 3.
Producer-5: Add one Fri Nov 25 15:36:26 CST 2016. Buffer size is 4.
Producer-6: Add one Fri Nov 25 15:36:27 CST 2016. Buffer size is 5.
Consumer-2: Get one Fri Nov 25 15:36:23 CST 2016. Buffer size is 4.
Producer-7: Add one Fri Nov 25 15:36:31 CST 2016. Buffer size is 5.
Consumer-3: Get one Fri Nov 25 15:36:24 CST 2016. Buffer size is 4.
Producer-8: Add one Fri Nov 25 15:36:36 CST 2016. Buffer size is 5.
Consumer-4: Get one Fri Nov 25 15:36:25 CST 2016. Buffer size is 4.
Producer-9: Add one Fri Nov 25 15:36:41 CST 2016. Buffer size is 5.
Consumer-5: Get one Fri Nov 25 15:36:26 CST 2016. Buffer size is 4.
Consumer-6: Get one Fri Nov 25 15:36:27 CST 2016. Buffer size is 3.
Consumer-7: Get one Fri Nov 25 15:36:31 CST 2016. Buffer size is 2.
Consumer-8: Get one Fri Nov 25 15:36:36 CST 2016. Buffer size is 1.
Consumer-9: Get one Fri Nov 25 15:36:41 CST 2016. Buffer size is 0.