【轉】-Java實現生產者和消費者的5種方式

Java實現生產者和消費者的5種方式

1. 前言

生產者和消費者問題是線程模型中的經典問題：生產者和消費者在同一時間段內共用同一個存儲空間，生產者往存儲空間中添加產品，消費者從存儲空間中取走產品，當存儲空間為空時，消費者阻塞，當存儲空間滿時，生產者阻塞。

現在用四種方式來實現生產者消費者模型

2. wait()和notify()方法的實現

這也是最簡單最基礎的實現，緩沖區滿和為空時都調用wait()方法等待，當生產者生產了一個產品或者消費者消費了一個產品之后會喚醒所有線程。

/**
 * 生產者和消費者，wait()和notify()的實現
 * @author ZGJ
 * @date 2017年6月22日
 */
public class Test1 {
    private static Integer count = 0;
    private static final Integer FULL = 10;
    private static String LOCK = "lock";
    
    public static void main(String[] args) {
        Test1 test1 = new Test1();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count == FULL) {
                        try {
                            LOCK.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName() + "生產者生產，目前總共有" + count);
                    LOCK.notifyAll();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count == 0) {
                        try {
                            LOCK.wait();
                        } catch (Exception e) {
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName() + "消費者消費，目前總共有" + count);
                    LOCK.notifyAll();
                }
            }
        }
    }
}

結果:

Thread-0生產者生產，目前總共有1
Thread-4生產者生產，目前總共有2
Thread-3消費者消費，目前總共有1
Thread-1消費者消費，目前總共有0
Thread-2生產者生產，目前總共有1
Thread-6生產者生產，目前總共有2
Thread-7消費者消費，目前總共有1
Thread-5消費者消費，目前總共有0
Thread-0生產者生產，目前總共有1
Thread-4生產者生產，目前總共有2
Thread-3消費者消費，目前總共有1
Thread-6生產者生產，目前總共有2
Thread-1消費者消費，目前總共有1
Thread-7消費者消費，目前總共有0
Thread-2生產者生產，目前總共有1
Thread-5消費者消費，目前總共有0
Thread-0生產者生產，目前總共有1
Thread-4生產者生產，目前總共有2
Thread-3消費者消費，目前總共有1
Thread-7消費者消費，目前總共有0
Thread-6生產者生產，目前總共有1
Thread-2生產者生產，目前總共有2
Thread-1消費者消費，目前總共有1
Thread-5消費者消費，目前總共有0
Thread-0生產者生產，目前總共有1
Thread-4生產者生產，目前總共有2
Thread-3消費者消費，目前總共有1
Thread-1消費者消費，目前總共有0
Thread-6生產者生產，目前總共有1
Thread-7消費者消費，目前總共有0
Thread-2生產者生產，目前總共有1

3. 可重入鎖ReentrantLock的實現

java.util.concurrent.lock 中的 Lock 框架是鎖定的一個抽象，通過對lock的lock()方法和unlock()方法實現了對鎖的顯示控制，而synchronize()則是對鎖的隱性控制。

可重入鎖，也叫做遞歸鎖，指的是同一線程外層函數獲得鎖之后，內層遞歸函數仍然有獲取該鎖的代碼，但不受影響，簡單來說，該鎖維護這一個與獲取鎖相關的計數器，如果擁有鎖的某個線程再次得到鎖，那么獲取計數器就加1，函數調用結束計數器就減1，然后鎖需要被釋放兩次才能獲得真正釋放。已經獲取鎖的線程進入其他需要相同鎖的同步代碼塊不會被阻塞。

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 生產者和消費者，ReentrantLock的實現
 * 
 * @author ZGJ
 * @date 2017年6月22日
 */
public class Test2 {
    private static Integer count = 0;
    private static final Integer FULL = 10;
    //創建一個鎖對象
    private Lock lock = new ReentrantLock();
    //創建兩個條件變量，一個為緩沖區非滿，一個為緩沖區非空
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    public static void main(String[] args) {
        Test2 test2 = new Test2();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                //獲取鎖
                lock.lock();
                try {
                    while (count == FULL) {
                        try {
                            notFull.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生產者生產，目前總共有" + count);
                    //喚醒消費者
                    notEmpty.signal();
                } finally {
                    //釋放鎖
                    lock.unlock();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                lock.lock();
                try {
                    while (count == 0) {
                        try {
                            notEmpty.await();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消費者消費，目前總共有" + count);
                    notFull.signal();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
}

4. 阻塞隊列BlockingQueue的實現

BlockingQueue即阻塞隊列，從阻塞這個詞可以看出，在某些情況下對阻塞隊列的訪問可能會造成阻塞。被阻塞的情況主要有如下兩種:

當隊列滿了的時候進行入隊列操作
當隊列空了的時候進行出隊列操作
因此，當一個線程對已經滿了的阻塞隊列進行入隊操作時會阻塞，除非有另外一個線程進行了出隊操作，當一個線程對一個空的阻塞隊列進行出隊操作時也會阻塞，除非有另外一個線程進行了入隊操作。
從上可知，阻塞隊列是線程安全的。
下面是BlockingQueue接口的一些方法:

操作	拋異常	特定值	阻塞	超時
插入	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
移除	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
檢查	element(o)	peek(o)

這四類方法分別對應的是：

ThrowsException：如果操作不能馬上進行，則拋出異常
SpecialValue：如果操作不能馬上進行，將會返回一個特殊的值，一般是true或者false
Blocks:如果操作不能馬上進行，操作會被阻塞
TimesOut:如果操作不能馬上進行，操作會被阻塞指定的時間，如果指定時間沒執行，則返回一個特殊值，一般是true或者false

下面來看由阻塞隊列實現的生產者消費者模型,這里我們使用take()和put()方法，這里生產者和生產者，消費者和消費者之間不存在同步，所以會出現連續生成和連續消費的現象

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * 使用BlockingQueue實現生產者消費者模型
 * @author ZGJ
 * @date 2017年6月29日
 */
public class Test3 {
    private static Integer count = 0;
    //創建一個阻塞隊列
    final BlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);
    public static void main(String[] args) {
        Test3 test3 = new Test3();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    blockingQueue.put(1);
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生產者生產，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    blockingQueue.take();
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消費者消費，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

5. 信號量Semaphore的實現

Semaphore（信號量）是用來控制同時訪問特定資源的線程數量，它通過協調各個線程，以保證合理的使用公共資源，在操作系統中是一個非常重要的問題，可以用來解決哲學家就餐問題。Java中的Semaphore維護了一個許可集，一開始先設定這個許可集的數量，可以使用acquire()方法獲得一個許可，當許可不足時會被阻塞，release()添加一個許可。在下列代碼中，還加入了另外一個mutex信號量，維護生產者消費者之間的同步關系，保證生產者和消費者之間的交替進行

import java.util.concurrent.Semaphore;
/**
 * 使用semaphore信號量實現
 * @author ZGJ
 * @date 2017年6月29日
 */
public class Test4 {
    private static Integer count = 0;
    //創建三個信號量
    final Semaphore notFull = new Semaphore(10);
    final Semaphore notEmpty = new Semaphore(0);
    final Semaphore mutex = new Semaphore(1);
    public static void main(String[] args) {
        Test4 test4 = new Test4();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    notFull.acquire();
                    mutex.acquire();
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生產者生產，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    mutex.release();
                    notEmpty.release();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    notEmpty.acquire();
                    mutex.acquire();
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消費者消費，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    mutex.release();
                    notFull.release();
                }
            }
        }
    }
}

6. 管道輸入輸出流PipedInputStream和PipedOutputStream實現

在java的io包下，PipedOutputStream和PipedInputStream分別是管道輸出流和管道輸入流。
它們的作用是讓多線程可以通過管道進行線程間的通訊。在使用管道通信時，必須將PipedOutputStream和PipedInputStream配套使用。

使用方法：先創建一個管道輸入流和管道輸出流，然后將輸入流和輸出流進行連接，用生產者線程往管道輸出流中寫入數據，消費者在管道輸入流中讀取數據，這樣就可以實現了不同線程間的相互通訊，但是這種方式在生產者和生產者、消費者和消費者之間不能保證同步，也就是說在一個生產者和一個消費者的情況下是可以生產者和消費者之間交替運行的，多個生成者和多個消費者者之間則不行

/**
 * 使用管道實現生產者消費者模型
 * @author ZGJ
 * @date 2017年6月30日
 */
public class Test5 {
    final PipedInputStream pis = new PipedInputStream();
    final PipedOutputStream pos = new PipedOutputStream();
    {
        try {
            pis.connect(pos);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            try {
                while(true) {
                    Thread.sleep(1000);
                    int num = (int) (Math.random() * 255);
                    System.out.println(Thread.currentThread().getName() + "生產者生產了一個數字，該數字為： " + num);
                    pos.write(num);
                    pos.flush();
                } 
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    pos.close();
                    pis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while(true) {
                    Thread.sleep(1000);
                    int num = pis.read();
                    System.out.println("消費者消費了一個數字，該數字為：" + num);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    pos.close();
                    pis.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) {
        Test5 test5 = new Test5();
        new Thread(test5.new Producer()).start();
        new Thread(test5.new Consumer()).start();
    }
}

posted @ 2024-07-10 09:44 booleandev 閱讀(29) 評論(0) 收藏舉報

刷新頁面返回頂部

Jantor的博客