Java實現PV操作 | 生產者與消費者

• 導語

在學習操作系統的過程中，PV操作是很重要的一個環節。然而面對書本上枯燥的代碼，每一個愛好技術的人總是想能親自去實現。現在我要推出一個專題，專門講述如何用Java實現PV操作，讓操作系統背后的邏輯躍然屏上。

如有錯誤，請廣大網友斧正，感激不盡！

---

經典問題1、生產者與消費者

 

• PV操作數據結構的構建

 

在書本上，我們給出了一種數據結構，叫做信號量。這種信號量有兩個元素：

　　一個是count，如果是正值則表示當前資源的個數，如果是0，表示有一個進程在執行臨界區的代碼（也就是說這個進程位於臨界區）；並且沒有進程處於阻塞隊列中。如果是負值，這個值的絕對值（abs(count)）表示阻塞隊列中進程的個數。

　　一個是queue，即為阻塞進程隊列。當進程不能申請相應的資源是，則使用P操作，將自己插入阻塞隊列中。當運行的進程執行完臨界區代碼時，就執行V操作，喚醒一個阻塞隊列中的進程。

 

現在我們定義一個PV操作類：syn。在這個類中我們可以通過構造函數設置count的值。可以看到這個類中並沒有阻塞進程所在的queue，我是通過java的this.wait()與this.notifyAll()來實現的。

並且，通過關鍵字【synchronized】，保證了PV操作是一條【原語】，即在運行過程中，占有完整的一個時間片，不可分割。

class syn{//PV操作類
    int count=0;//信號量
    syn(){}
    syn(int a){count=a;}
    public synchronized void Wait(){ //關鍵字 synchronized 保證了此操作是一條【原語】
        count--;
        if(count<0){//等於0 ：有一個進程進入了臨界區
            try {         //小於0：abs(count)=阻塞的進程數目
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();  
            }  
        }  
    }  
    public synchronized void Signal(){   //關鍵字 synchronized 保證了此操作是一條【原語】
        count++;
        if(count<=0){//如果有進程阻塞
            this.notify(); 19         }
    }  
}

 注1：P操作wait中，應為if(count<0)而不是while(count<0) 。在后續的編寫中，發現如果引入了多個生產者與消費者，用語句while(count<0) 就會出錯。

 注2：V操作signal中，應為this.notify()而不是this.notifyAll()。也就是說只需要從阻塞隊列中喚醒一個進程。

---

 

• 單個生產者與消費者的模型構建

我們構建實現了Runnable接口的生產者類和消費者類，使用empty（表示空緩沖區的數目）和full（表示滿緩沖區的數目）兩個信號量來實現進程的同步。（不同類型的進程共享某一資源為同步關系）

代碼如下：

 

public class Main {

    public static void main(String[] args) {
        Producer p=new Producer();
        Consumer c=new Consumer();
        Thread pp=new Thread(p);
        Thread cp=new Thread(c);
        pp.start();
        cp.start();
    }
}

class Global{
    static syn empty=new syn(8);
    static syn full=new syn(0);
    static int buffer []=new int[8];//緩沖區
}

//生產者類
class Producer implements Runnable{
    int count=0;
    public void run(){
        while(count<20){
            Global.empty.Wait();
            //臨界區
            int index=count%8;
            Global.buffer[index]=count;
            System.out.println("生產者在緩沖區"+index+"中生產了物品"+count);
            count++;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // end of 臨界區
            Global.full.Signal();
        }
    }
}

//消費者類
class Consumer implements Runnable{
    int count=0;
    public void run(){
        while(count<20){
            Global.full.Wait();
            //臨界區
            int index=count%8;
            int value=Global.buffer[index];
            System.out.println("消費者在緩沖區"+index+"中消費了物品"+value);
            count++;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // end of 臨界區
            Global.empty.Signal();
        }
    }
}

 

運行結果：

 

 可以看出進程嚴格按照先生產再消費的順序，完美運行。

---

 

• 引入多個生產者與消費者

為保證同一類進程在進行訪問時能保證互斥（互斥是同類進程共享某一資源時的方式）我們引入mutex信號量。

    static syn pMutex=new syn(1);//保證生產者之間互斥
    static syn cMutex=new syn(1);//保證消費者之間互斥

 運行結果：

 

可見程序完美運行。

---

完整Java代碼：

public class Main {

    public static void main(String[] args) {
        Producer p[]=new Producer[3];//3個生產者
        Consumer c[]=new Consumer[3];
        int i;
        
        for(i=0;i<3;i++){
            p[i]=new Producer(i+1);
        }
        for(i=0;i<3;i++){
            c[i]=new Consumer(i+1);
        }
        
        Thread pp[]=new Thread[3];
        Thread cp[]=new Thread[3];
        
        for(i=0;i<3;i++){
            pp[i]=new Thread(p[i]);
        }
        for(i=0;i<3;i++){
            cp[i]=new Thread(c[i]);
        }
        
        for(i=0;i<3;i++){
            pp[i].start();
        }
        for(i=0;i<3;i++){
            cp[i].start();
        }

    }
}

class Global{
    static syn empty=new syn(8);
    static syn full=new syn(0);
    static syn pMutex=new syn(1);//保證生產者之間互斥
    static syn cMutex=new syn(1);//保證消費者之間互斥
    static int buffer []=new int[8];//緩沖區
    static int pCount=0;
    static int cCount=0;
}

//生產者類
class Producer implements Runnable{
    int ID=0;
    Producer(){}
    Producer(int id){ID=id;}
    public void run(){
        while(Global.pCount<20){
            Global.empty.Wait();
            Global.pMutex.Wait();
            //臨界區
            int index=Global.pCount%8;
            Global.buffer[index]=Global.pCount;
            System.out.println("生產者"+ID+"在緩沖區"+index+"中生產了物品"+Global.pCount);
            Global.pCount++;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // end of 臨界區
            Global.pMutex.Signal();
            Global.full.Signal();
        }
    }
}

//消費者類
class Consumer implements Runnable{
    int ID=0;
    Consumer(){}
    Consumer(int id){ID=id;}
    public void run(){
        while(Global.cCount<20){
            Global.full.Wait();
            Global.cMutex.Wait();
            //臨界區
            int index=Global.cCount%8;
            int value=Global.buffer[index];
            System.out.println("消費者"+ID+"在緩沖區"+index+"中消費了物品"+value);
            Global.cCount++;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // end of 臨界區
            Global.cMutex.Signal();
            Global.empty.Signal();
        }
    }
}

class syn{//PV操作類
    int count=0;//信號量
    syn(){}
    syn(int a){count=a;}
    public synchronized void Wait(){ //關鍵字 synchronized 保證了此操作是一條【原語】
        count--;
        if(count<0){//等於0 ：有一個進程進入了臨界區
            try {         //小於0：abs(count)=阻塞的進程數目
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();  
            }  
        }  
    }  
    public synchronized void Signal(){   //關鍵字 synchronized 保證了此操作是一條【原語】
        count++;
        if(count<=0){//如果有進程阻塞
            this.notify();//All
        }
    }  
}