實現生產者消費者模式的四種方式（Synchronized、Lock、Semaphore、Blocki

godiscoder 發布于2019-08-19 11:05 / 3065人閱讀

摘要：所謂生產者消費者模式，即個線程進行生產，同時個線程進行消費，兩種角色通過內存緩沖區進行通信圖片來源下面我們通過四種方式，來實現生產者消費者模式。通過生產者調用，減少數目可以消費的數量。

所謂生產者消費者模式，即N個線程進行生產，同時N個線程進行消費，兩種角色通過內存緩沖區進行通信

圖片來源https://www.cnblogs.com/chent...

下面我們通過四種方式，來實現生產者消費者模式。

首先是最原始的synchronized方式

定義庫存類（即圖中緩存區）

class Stock {
    private String name;
    // 標記庫存是否有內容
    private boolean hasComputer = false;

    public synchronized void putOne(String name) {
        // 若庫存中已有內容，則生產線程阻塞等待
        while (hasComputer) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.name = name;
        System.out.println("生產者...生產了 " + name);
        // 更新標記
        this.hasComputer = true;
        // 這里用notify的話，假設p0執行完畢，此時c0,c1都在wait, 同時喚醒另一個provider:p1，
        // p1判斷標記后休眠，造成所有線程都wait的局面，即死鎖；
        // 因此使用notifyAll解決死鎖問題
        this.notifyAll();
    }

    public synchronized void takeOne() {
        // 若庫存中沒有內容，則消費線程阻塞等待生產完畢后繼續
        while (!hasComputer) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("消費者...消費了 " + name);
        this.hasComputer = false;
        this.notifyAll();
    }
}

定義生產者和消費者（為了節省空間和方便閱讀，這里將生產者和消費者定義成了匿名內部類）

public static void main(String[] args) {
    // 用于通信的庫存類
    Stock computer = new Stock();
    // 定義兩個生產者和兩個消費者
    Thread p1 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.putOne("Dell");
            }
        }
    });
    Thread p2 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.putOne("Mac");
            }
        }
    });
    
    Thread c1 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.takeOne();
            }
        }
    });
    Thread c2 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.takeOne();
            }
        }
    });
    p1.start();
    p2.start();
    c1.start();
    c2.start();
}

運行結果圖

第二種方式：Lock

Jdk1.5之后加入了Lock接口，一個lock對象可以有多個Condition類,Condition類負責對lock對象進行wait,notify,notifyall操作

定義庫存類

class LockStock {
    final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    // 加入庫存概念，可批量生產和消費
    // 定義最大庫存為10
    final String[] stock = new String[10];
    // 寫入標記、讀取標記、已有商品數量
    int putptr, takeptr, count;

    public void put(String computer) {
        // lock代替synchronized
        lock.lock();
        try {
            // 若庫存已滿則生產者線程阻塞
            while (count == stock.length)
                notFull.await();
            // 庫存中加入商品
            stock[putptr] = computer;
            // 庫存已滿，指針置零，方便下次重新寫入
            if (++putptr == stock.length) putptr = 0;
            ++count;
            System.out.println(computer + " 正在生產數據： -- 庫存剩余：" + count);
            notEmpty.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String take(String consumerName) {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            // 從庫存中獲取商品
            String computer = stock[takeptr];
            if (++takeptr == stock.length) takeptr = 0;
            --count;
            System.out.println(consumerName + " 正在消費數據：" + computer + " -- 庫存剩余：" + count);
            notFull.signal();
            return computer;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

        // 無邏輯作用，放慢速度
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "";
    }
}

以上部分代碼摘自java7 API中Condition接口的官方示例

接著還是定義生產者和消費者

public static void main(String[] args) {
    LockStock computer = new LockStock();
    Thread p1 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.put("Dell");
            }
        }
    });
    Thread p2 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.put("Mac");
            }
        }
    });

    Thread c1 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.take("zhangsan");
            }
        }
    });
    Thread c2 = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                computer.take("李四");
            }
        }
    });
    // 兩個生產者兩個消費者同時運行
    p1.start();
    p2.start();
    c1.start();
    c2.start();
}

運行結果圖：

第三種方式：Semaphore
首先依舊是庫存類：

class Stock {
    List stock = new LinkedList();
    // 互斥量，控制共享數據的互斥訪問
    private Semaphore mutex = new Semaphore(1);

    // canProduceCount可以生產的總數量。 通過生產者調用acquire，減少permit數目
    private Semaphore canProduceCount = new Semaphore(10);

    // canConsumerCount可以消費的數量。通過生產者調用release，增加permit數目
    private Semaphore canConsumerCount = new Semaphore(0);

    public void put(String computer) {
        try {
            // 可生產數量 -1
            canProduceCount.acquire();
            mutex.acquire();
            // 生產一臺電腦
            stock.add(computer);
            System.out.println(computer + " 正在生產數據" + " -- 庫存剩余：" + stock.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 釋放互斥鎖
            mutex.release();
            // 釋放canConsumerCount，增加可以消費的數量
            canConsumerCount.release();
        }
        // 無邏輯作用，放慢速度
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void get(String consumerName) {
        try {
            // 可消費數量 -1
            canConsumerCount.acquire();
            mutex.acquire();
            // 從庫存消費一臺電腦
            String removedVal = stock.remove(0);
            System.out.println(consumerName + " 正在消費數據：" + removedVal + " -- 庫存剩余：" + stock.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            mutex.release();
            // 消費后釋放canProduceCount，增加可以生產的數量
            canProduceCount.release();
        }
    }
}

還是生產消費者：

public class SemaphoreTest {
    public static void main(String[] args) {
        // 用于多線程操作的庫存變量
        final Stock stock = new Stock();
        // 定義兩個生產者和兩個消費者
        Thread dellProducer = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    stock.put("Del");
                }
            }
        });
        Thread macProducer = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    stock.put("Mac");
                }
            }
        });
        Thread consumer1 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    stock.get("zhangsan");
                }
            }
        });
        Thread consumer2 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    stock.get("李四");
                }
            }
        });
        dellProducer.start();
        macProducer.start();
        consumer1.start();
        consumer2.start();
    }
}

運行結果圖：

第四種方式：BlockingQueue
BlockingQueue的put和take底層實現其實也是使用了第二種方式中的ReentrantLock+Condition，并且幫我們實現了庫存隊列，方便簡潔
1、定義生產者

class Producer implements Runnable {
    // 庫存隊列
    private BlockingQueue stock;
    // 生產/消費延遲
    private int timeOut;
    private String name;

    public Producer(BlockingQueue stock, int timeout, String name) {
        this.stock = stock;
        this.timeOut = timeout;
        this.name = name;
    }

    @Override
    public void run() {
        while (true) {
            try {
                stock.put(name);
                System.out.println(name + " 正在生產數據" + " -- 庫存剩余：" + stock.size());
                TimeUnit.MILLISECONDS.sleep(timeOut);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

2、定義消費者

class Consumer implements Runnable {
    // 庫存隊列
    private BlockingQueue stock;
    private String consumerName;

    public Consumer(BlockingQueue stock, String name) {
        this.stock = stock;
        this.consumerName = name;
    }

    @Override
    public void run() {
        while (true) {
            try {
                // 從庫存消費一臺電腦
                String takeName = stock.take();
                System.out.println(consumerName + " 正在消費數據：" + takeName + " -- 庫存剩余：" + stock.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

3、定義庫存并運行

public static void main(String[] args) {
        // 定義最大庫存為10
        BlockingQueue stock = new ArrayBlockingQueue<>(10);
        Thread p1 = new Thread(new Producer(stock, 500, "Mac"));
        Thread p2 = new Thread(new Producer(stock, 500, "Dell"));
        Thread c1 = new Thread(new Consumer(stock,"zhangsan"));
        Thread c2 = new Thread(new Consumer(stock, "李四"));

        p1.start();
        p2.start();
        c1.start();
        c2.start();

    }

運行結果圖：

感謝閱讀~歡迎指正和補充~~~

GPU云服務器云服務器對數據倉庫分析的四種方式對集群生產消費實現數據遷移 activity的四種狀態數據遷移的四種方法

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/76979.html

Java 中的 Monitor 機制

摘要：基本元素機制需要幾個元素來配合，分別是臨界區對象及鎖條件變量以及定義在對象上的，操作。這個外部條件在機制中稱為條件變量。提供的機制，其實是，等元素合作形成的，甚至說外部的條件變量也是個組成部分。 monitor的概念管程，英文是 Monitor，也常被翻譯為監視器，monitor 不管是翻譯為管程還是監視器，都是比較晦澀的，通過翻譯后的中文，并無法對 monitor 達到一個直觀的描...

Jacendfeng 2019-08-19 11:10 評論0 收藏0
想進大廠？50個多線程面試題，你會多少？（一）

摘要：下面是線程相關的熱門面試題，你可以用它來好好準備面試。線程安全問題都是由全局變量及靜態變量引起的。持有自旋鎖的線程在之前應該釋放自旋鎖以便其它線程可以獲得自旋鎖。最近看到網上流傳著，各種面試經驗及面試題，往往都是一大堆技術題目貼上去，而沒有答案。不管你是新程序員還是老手，你一定在面試中遇到過有關線程的問題。Java語言一個重要的特點就是內置了對并發的支持，讓Java大受企業和程序員...

wow_worktile 2019-08-15 14:50 評論0 收藏0
40道阿里巴巴JAVA研發崗多線程面試題詳解，你能答出多少

摘要：但是單核我們還是要應用多線程，就是為了防止阻塞。多線程可以防止這個問題，多條線程同時運行，哪怕一條線程的代碼執行讀取數據阻塞，也不會影響其它任務的執行。 1、多線程有什么用？一個可能在很多人看來很扯淡的一個問題：我會用多線程就好了，還管它有什么用？在我看來，這個回答更扯淡。所謂知其然知其所以然，會用只是知其然，為什么用才是知其所以然，只有達到知其然知其所以然的程度才可以說是把一個知識點...

lpjustdoit 2019-08-16 18:36 評論0 收藏0
Java 并發學習筆記

摘要：方法可以將當前線程放入等待集合中，并釋放當前線程持有的鎖。此后，該線程不會接收到的調度，并進入休眠狀態。該線程會喚醒，并嘗試恢復之前的狀態。并發最近重新復習了一邊并發的知識，發現自己之前對于并發的了解只是皮毛。這里總結以下Java并發需要掌握的點。使用并發的一個重要原因是提高執行效率。由于I/O等情況阻塞，單個任務并不能充分利用CPU時間。所以在單處理器的機器上也應該使用并發。為...

DrizzleX 2019-08-14 14:14 評論0 收藏0