java中的wait()和notify()方法實現(xiàn)生產(chǎn)者消費者模式實例

這篇文章主要介紹“java中的wait()和notify()方法實現(xiàn)生產(chǎn)者消費者模式實例”，在日常操作中，相信很多人在java中的wait()和notify()方法實現(xiàn)生產(chǎn)者消費者模式實例問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”java中的wait()和notify()方法實現(xiàn)生產(chǎn)者消費者模式實例”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

java wait()/notify() 實現(xiàn)生產(chǎn)者消費者模式

java中的多線程會涉及到線程間通信，常見的線程通信方式，例如共享變量、管道流等，這里我們要實現(xiàn)生產(chǎn)者消費者模式，也需要涉及到線程通信，不過這里我們用到了java中的wait()、notify()方法：

wait()：進(jìn)入臨界區(qū)的線程在運行到一部分后，發(fā)現(xiàn)進(jìn)行后面的任務(wù)所需的資源還沒有準(zhǔn)備充分，所以調(diào)用wait()方法，讓線程阻塞，等待資源，同時釋放臨界區(qū)的鎖，此時線程的狀態(tài)也從RUNNABLE狀態(tài)變?yōu)閃AITING狀態(tài)；

notify()：準(zhǔn)備資源的線程在準(zhǔn)備好資源后，調(diào)用notify()方法通知需要使用資源的線程，同時釋放臨界區(qū)的鎖，將臨界區(qū)的鎖交給使用資源的線程。

wait()、notify()這兩個方法，都必須要在臨界區(qū)中調(diào)用，即是在synchronized同步塊中調(diào)用，不然會拋出IllegalMonitorStateException的異常。

實現(xiàn)源碼：

生產(chǎn)者線程類：

package threads; 
import java.util.List;
import java.util.UUID; 
public class Producer extends Thread{ 
 private List<String> storage;//生產(chǎn)者倉庫
 public Producer(List<String> storage) {
  this.storage = storage;
 }
 public void run(){
  //生產(chǎn)者每隔1s生產(chǎn)1~100消息
  long oldTime = System.currentTimeMillis();
  while(true){
   synchronized(storage){
    if (System.currentTimeMillis() - oldTime >= 1000) {
     oldTime = System.currentTimeMillis();
     int size = (int)(Math.random()*100) + 1;
     for (int i = 0; i < size; i++) {
      String msg = UUID.randomUUID().toString();
      storage.add(msg);
     }
     System.out.println("線程"+this.getName()+"生產(chǎn)消息"+size+"條");
     storage.notify();
    }
   }
  }
 }
}

消費者線程類：

package threads; 
import java.util.List; 
public class Consumer extends Thread{ 
 private List<String> storage;//倉庫
 public Consumer(List<String> storage) {
  this.storage = storage;
 }
 public void run(){
  while(true){
   synchronized(storage){
    //消費者去倉庫拿消息的時候，如果發(fā)現(xiàn)倉庫數(shù)據(jù)為空，則等待
    if (storage.isEmpty()) {
     try {
      storage.wait();
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
    int size = storage.size();
    for (int i = size - 1; i >= 0; i--) {
     storage.remove(i);
    }
    System.out.println("線程"+this.getName()+"成功消費"+size+"條消息");
   }
  }
 }
}

倉庫類：

package threads; 
import java.util.ArrayList;
import java.util.List; 

public class Storage { 
 private List<String> storage;//生產(chǎn)者和消費者共享的倉庫
 public Storage() {
  storage = new ArrayList<String>();
 }
 public List<String> getStorage() {
  return storage;
 }
 public void setStorage(List<String> storage) {
  this.storage = storage;
 } 
}

main方法類：

package threads; 
public class App { 
 public static void main(String[] args) {
  Storage storage = new Storage();
  Producer producer = new Producer(storage.getStorage());
  Consumer consumer = new Consumer(storage.getStorage());
  producer.start();
  consumer.start();
 }
}

生產(chǎn)消費效果：

Wait/Notify通知機制解析

前言

我們知道，java的wait/notify的通知機制可以用來實現(xiàn)線程間通信。wait表示線程的等待，調(diào)用該方法會導(dǎo)致線程阻塞，直至另一線程調(diào)用notify或notifyAll方法才可另其繼續(xù)執(zhí)行。經(jīng)典的生產(chǎn)者、消費者模式即是使用wait/notify機制得以完成。在這篇文章中，我們將深入解析這一機制，了解其背后的原理。

線程的狀態(tài)

在了解wait/notify機制前，先熟悉一下java線程的幾個生命周期。分別為初始（NEW）、運行(RUNNABLE)、阻塞(BLOCKED)、等待(WAITING)、超時等待(TIMED_WAITING)、終止(TERMINATED)等狀態(tài)（位于java.lang.Thread.State枚舉類中）。

以下是對這幾個狀態(tài)的簡要說明，詳細(xì)說明見該類注釋。

對于以上線程間的狀態(tài)及轉(zhuǎn)化關(guān)系，我們需要知道

• WAITING(等待狀態(tài))和TIMED_WAITING(超時等待)都會令線程進(jìn)入等待狀態(tài)，不同的是TIMED_WAITING會在超時后自行返回，而WAITING則需要等待至條件改變。

• 進(jìn)入阻塞狀態(tài)的唯一前提是在等待獲取同步鎖。java注釋說的很明白，只有兩種情況可以使線程進(jìn)入阻塞狀態(tài)：一是等待進(jìn)入synchronized塊或方法，另一個是在調(diào)用wait()方法后重新進(jìn)入synchronized塊或方法。下文會有詳細(xì)解釋。

• Lock類對于鎖的實現(xiàn)不會令線程進(jìn)入阻塞狀態(tài)，Lock底層調(diào)用LockSupport.park()方法，使線程進(jìn)入的是等待狀態(tài)。

wait/notify用例

讓我們先通過一個示例解析

wait()方法可以使線程進(jìn)入等待狀態(tài)，而notify()可以使等待的狀態(tài)喚醒。這樣的同步機制十分適合生產(chǎn)者、消費者模式：消費者消費某個資源，而生產(chǎn)者生產(chǎn)該資源。當(dāng)該資源缺失時，消費者調(diào)用wait()方法進(jìn)行自我阻塞，等待生產(chǎn)者的生產(chǎn)；生產(chǎn)者生產(chǎn)完畢后調(diào)用notify/notifyAll()喚醒消費者進(jìn)行消費。

以下是代碼示例，其中flag標(biāo)志表示資源的有無。

public class ThreadTest {
    static final Object obj = new Object();
    private static boolean flag = false;
    public static void main(String[] args) throws Exception {
        Thread consume = new Thread(new Consume(), "Consume");
        Thread produce = new Thread(new Produce(), "Produce");
        consume.start();
        Thread.sleep(1000);
        produce.start();
        try {
            produce.join();
            consume.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    // 生產(chǎn)者線程
    static class Produce implements Runnable {
        @Override
        public void run() {
            synchronized (obj) {
                System.out.println("進(jìn)入生產(chǎn)者線程");
                System.out.println("生產(chǎn)");
                try {
                    TimeUnit.MILLISECONDS.sleep(2000);  //模擬生產(chǎn)過程
                    flag = true;
                    obj.notify();  //通知消費者
                    TimeUnit.MILLISECONDS.sleep(1000);  //模擬其他耗時操作
                    System.out.println("退出生產(chǎn)者線程");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    //消費者線程
    static class Consume implements Runnable {
        @Override
        public void run() {
            synchronized (obj) {
                System.out.println("進(jìn)入消費者線程");
                System.out.println("wait flag 1:" + flag);
                while (!flag) {  //判斷條件是否滿足，若不滿足則等待
                    try {
                        System.out.println("還沒生產(chǎn)，進(jìn)入等待");
                        obj.wait();
                        System.out.println("結(jié)束等待");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("wait flag 2:" + flag);
                System.out.println("消費");
                System.out.println("退出消費者線程");
            }
        }
    }
}

輸出結(jié)果為：

進(jìn)入消費者線程

wait flag 1:false

還沒生產(chǎn)，進(jìn)入等待

進(jìn)入生產(chǎn)者線程

生產(chǎn)

退出生產(chǎn)者線程

結(jié)束等待

wait flag 2:true

消費

退出消費者線程

理解了輸出結(jié)果的順序，也就明白了wait/notify的基本用法。有以下幾點需要知道：

• 在示例中沒有體現(xiàn)但很重要的是，wait/notify方法的調(diào)用必須處在該對象的鎖（Monitor）中，也即，在調(diào)用這些方法時首先需要獲得該對象的鎖。否則會爬出IllegalMonitorStateException異常。

• 從輸出結(jié)果來看，在生產(chǎn)者調(diào)用notify()后，消費者并沒有立即被喚醒，而是等到生產(chǎn)者退出同步塊后才喚醒執(zhí)行。（這點其實也好理解，synchronized同步方法（塊）同一時刻只允許一個線程在里面，生產(chǎn)者不退出，消費者也進(jìn)不去）

• 注意，消費者被喚醒后是從wait()方法（被阻塞的地方）后面執(zhí)行，而不是重新從同步塊開頭。

深入了解

這一節(jié)我們探討wait/notify與線程狀態(tài)之間的關(guān)系。深入了解線程的生命周期。

由前面線程的狀態(tài)轉(zhuǎn)化圖可知，當(dāng)調(diào)用wait()方法后，線程會進(jìn)入WAITING(等待狀態(tài))，后續(xù)被notify()后，并沒有立即被執(zhí)行，而是進(jìn)入等待獲取鎖的阻塞隊列。

對于每個對象來說，都有自己的等待隊列和阻塞隊列。以前面的生產(chǎn)者、消費者為例，我們拿obj對象作為對象鎖，配合圖示。內(nèi)部流程如下

• 當(dāng)線程A（消費者）調(diào)用wait()方法后，線程A讓出鎖，自己進(jìn)入等待狀態(tài)，同時加入鎖對象的等待隊列。

• 線程B（生產(chǎn)者）獲取鎖后，調(diào)用notify方法通知鎖對象的等待隊列，使得線程A從等待隊列進(jìn)入阻塞隊列。

• 線程A進(jìn)入阻塞隊列后，直至線程B釋放鎖后，線程A競爭得到鎖繼續(xù)從wait()方法后執(zhí)行。

到此，關(guān)于“java中的wait()和notify()方法實現(xiàn)生產(chǎn)者消費者模式實例”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>