Java实现PV操作 | 读者与写者（在三种情况下进行讨论）

注 ：本文应结合【天勤笔记】进行学习。

1.读者优先

设置rmutex信号量来对readcount变量进行互斥访问、mutex信号量对写者与读者进行同步。

    static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
    static syn mutex=new syn(1);//多个【写者】对数据区进行【互斥】访问

java代码：（点击加号可查看）

package 读者优先;
import java.util.Scanner;

public class Main {

    public static void main(String[] args) {
        System.out.print("请设置读者数目:");
        Scanner scan=new Scanner(System.in);
        int readNum =scan.nextInt();
        
        System.out.print("请设置写者数目:");
        scan=new Scanner(System.in);
        int writeNum =scan.nextInt();        
        
        System.out.print("请设置循环上限:");
        scan=new Scanner(System.in);
        Global.UpBound =scan.nextInt();
        
        scan.close();

        
        Reader r[]=new Reader[readNum];
        Writer w[]=new Writer[writeNum];
        int i;
        for(i=0;i<readNum;i++){
            r[i]=new Reader(i+1);
        }
        for(i=0;i<writeNum;i++){
            w[i]=new Writer(i+1);
        }
        Thread []r_t=new Thread[readNum];
        Thread []w_t=new Thread[writeNum];
        for(i=0;i<readNum;i++){
            r_t[i]=new Thread(r[i]);
        }        
        for(i=0;i<writeNum;i++){
            w_t[i]=new Thread(w[i]);
        }
        for(i=0;i<writeNum;i++){
            w_t[i].start();
        }    
        for(i=0;i<readNum;i++){
            r_t[i].start();
        }    
        

    
    }

}

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
        }
    }  
}

class Global{
    static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
    static syn mutex=new syn(1);//多个【写者】对数据区进行【互斥】访问
    static int dataZone=0;    //数据区
    static int readcount=0;    //用于记录读者的数量
    static int data=0;
    static int UpBound=20;
}

class Reader implements Runnable{//读者
    int ID=0;
    Reader(){}
    Reader(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            //对readcount进行操作
            Global.rmutex.Wait();
            if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
                Global.mutex.Wait();
            }
            Global.readcount++;//读者数量增减
            Global.rmutex.Signal();
            //对readcount操作结束
            
            /*
             * 进行读操作
             */
            int readData=Global.dataZone;
            System.out.println("读者"+ID+"读出了数据:"+readData);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            /*
             * 结束读操作
             */
            
            //对readcount进行操作
            Global.rmutex.Wait();
            Global.readcount--;//读者数量减少
            if(Global.readcount==0){//这是最后一个读者，唤醒写者
                Global.mutex.Signal();
            }
            Global.rmutex.Signal();
            //对readcount操作结束
        }
    }
}

class Writer implements Runnable{//写者
    int ID=0;
    Writer(){}
    Writer(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            Global.mutex.Wait();    //申请对数据区进行访问
            /*
             * 进行写操作
             */
            Global.data++;
            int writeData=Global.data;
            System.out.println("写者"+ID+"写入了数据:"+writeData);
            Global.dataZone=Global.data;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            /*
             * 结束写操作
             */
            Global.mutex.Signal();    //释放数据区，允许其他进程读写
        }
    }
}

---

2.公平策略

在这里，增加wmutex信号量来表示是否有正在进行或等待的写者：

static syn wmutex=new syn(1);//表示是否存在【进行】或【等待】的【写者】

在读者readcount进入区和离开区增加wait（wmutex）和signal（wmutex）的操作：

在写者的进入区与离开区增加wait（wmutex）和signal（wmutex）的操作：

java代码：（点击加号可查看）

package 公平策略;

import java.util.Scanner;



public class Main {

    public static void main(String[] args) {
        System.out.print("请设置读者数目:");
        Scanner scan=new Scanner(System.in);
        int readNum =scan.nextInt();
        
        System.out.print("请设置写者数目:");
        scan=new Scanner(System.in);
        int writeNum =scan.nextInt();        
        
        System.out.print("请设置循环上限:");
        scan=new Scanner(System.in);
        Global.UpBound =scan.nextInt();
        
        scan.close();

        
        Reader r[]=new Reader[readNum];
        Writer w[]=new Writer[writeNum];
        int i;
        for(i=0;i<readNum;i++){
            r[i]=new Reader(i+1);
        }
        for(i=0;i<writeNum;i++){
            w[i]=new Writer(i+1);
        }
        Thread []r_t=new Thread[readNum];
        Thread []w_t=new Thread[writeNum];
        for(i=0;i<readNum;i++){
            r_t[i]=new Thread(r[i]);
        }        
        for(i=0;i<writeNum;i++){
            w_t[i]=new Thread(w[i]);
        }
        for(i=0;i<writeNum;i++){
            w_t[i].start();
        }    
        for(i=0;i<readNum;i++){
            r_t[i].start();
        }    
        

    
    }

}

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
        }
    }  
}

class Global{
    static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
    static syn mutex=new syn(1);//多个【写者】对数据区进行【互斥】访问
    static int dataZone=0;    //数据区
    static int readcount=0;    //用于记录读者的数量
    static int data=0;
    static int UpBound=20;
    static syn wmutex=new syn(1);//表示是否存在【进行】或【等待】的【写者】
}

class Reader implements Runnable{//读者
    int ID=0;
    Reader(){}
    Reader(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            Global.wmutex.Wait();//检测是否存在写者，无写者才能进入
            
            //对readcount进行操作
            Global.rmutex.Wait();
            if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
                Global.mutex.Wait();
            }
            Global.readcount++;//读者数量增减
            Global.rmutex.Signal();
            //对readcount操作结束
            
            Global.wmutex.Signal();//恢复wmutex
            
            /*
             * 进行读操作
             */
            int readData=Global.dataZone;
            System.out.println("读者"+ID+"读出了数据:"+readData);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            /*
             * 结束读操作
             */
            
            //对readcount进行操作
            Global.rmutex.Wait();
            Global.readcount--;//读者数量减少
            if(Global.readcount==0){//这是最后一个读者，唤醒写者
                Global.mutex.Signal();
            }
            Global.rmutex.Signal();
            //对readcount操作结束
            
            
        }
    }
}

class Writer implements Runnable{//写者
    int ID=0;
    Writer(){}
    Writer(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            Global.wmutex.Wait();
            Global.mutex.Wait();    //申请对数据区进行访问
            /*
             * 进行写操作
             */
            Global.data++;
            int writeData=Global.data;
            System.out.println("写者"+ID+"写入了数据:"+writeData);
            Global.dataZone=Global.data;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            /*
             * 结束写操作
             */
            Global.mutex.Signal();    //释放数据区，允许其他进程读写
            Global.wmutex.Signal();
        }
    }
}

---

3.写者优先

 公平策略是在读者优先的基础上进行修改，写者优先也是在公平策略的基础上进行修改。

在这里，我们增加了readable信号量，writecount全局变量。

在读者中，用readable代替了【公平策略】中的wmutex来对等待队列中的写者进行标记：

在写者中，通过判断等待队列中是否有写者，来控制读者的进入，并用wmutex对writecount全局变量进行互斥访问：

java代码：（点击加号可查看）

package 写者优先;

import java.util.Scanner;



public class Main {

    public static void main(String[] args) {
        System.out.print("请设置读者数目:");
        Scanner scan=new Scanner(System.in);
        int readNum =scan.nextInt();
        
        System.out.print("请设置写者数目:");
        scan=new Scanner(System.in);
        int writeNum =scan.nextInt();        
        
        System.out.print("请设置循环上限:");
        scan=new Scanner(System.in);
        Global.UpBound =scan.nextInt();
        
        scan.close();

        
        Reader r[]=new Reader[readNum];
        Writer w[]=new Writer[writeNum];
        int i;
        for(i=0;i<readNum;i++){
            r[i]=new Reader(i+1);
        }
        for(i=0;i<writeNum;i++){
            w[i]=new Writer(i+1);
        }
        Thread []r_t=new Thread[readNum];
        Thread []w_t=new Thread[writeNum];
        for(i=0;i<readNum;i++){
            r_t[i]=new Thread(r[i]);
        }        
        for(i=0;i<writeNum;i++){
            w_t[i]=new Thread(w[i]);
        }
        for(i=0;i<writeNum;i++){
            w_t[i].start();
        }    
        for(i=0;i<readNum;i++){
            r_t[i].start();
        }    
        

    
    }

}

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
        }
    }  
}

class Global{
    static syn mutex=new syn(1);//控制互斥访问的数据区    
    static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
    static syn wmutex=new syn(1);//多个【写者】对writecount进行【互斥】访问
    static syn readable=new syn(1);//表示当前是否有写者
    
    static int dataZone=0;    //数据区
    static int readcount=0;    //用于记录读者的数量
    static int writecount=0;    //用于记录读者的数量
    
    static int data=0;
    static int UpBound=20;

}

class Reader implements Runnable{//读者
    int ID=0;
    Reader(){}
    Reader(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            Global.readable.Wait();//检测是否存在写者，无写者才能进入
            
            //对readcount进行操作
            Global.rmutex.Wait();
            if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
                Global.mutex.Wait();
            }
            Global.readcount++;//读者数量增减
            Global.rmutex.Signal();
            //对readcount操作结束
            
            Global.readable.Signal();//恢复readable
            
            /*
             * 进行读操作
             */
            int readData=Global.dataZone;
            System.out.println("读者"+ID+"读出了数据:"+readData);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            /*
             * 结束读操作
             */
            
            //对readcount进行操作
            Global.rmutex.Wait();
            Global.readcount--;//读者数量减少
            if(Global.readcount==0){//这是最后一个读者，唤醒写者
                Global.mutex.Signal();
            }
            Global.rmutex.Signal();
            //对readcount操作结束
        }
    }
}

class Writer implements Runnable{//写者
    int ID=0;
    Writer(){}
    Writer(int id){ID=id;}
    public void run(){
        while(Global.data<=Global.UpBound){
            Global.wmutex.Wait();//准备修改writecount
            if(Global.writecount==0) Global.readable.Wait();//如果是第一个读者，则阻止后续读者进入
            Global.writecount++;
            Global.wmutex.Signal();//结束对writecount的修改
            
            Global.mutex.Wait();    //申请对数据区进行访问
            /*
             * 进行写操作
             */
            Global.data++;
            int writeData=Global.data;
            System.out.println("写者"+ID+"写入了数据:"+writeData);
            Global.dataZone=Global.data;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            /*
             * 结束写操作
             */
            Global.mutex.Signal();    //释放数据区，允许其他进程读写

            Global.wmutex.Wait();//准备修改writecount
            Global.writecount--;            
            if(Global.writecount==0) Global.readable.Signal();//如果是最后一个写者，唤醒读者
            Global.wmutex.Signal();//结束对writecount的修改
        }
    }
}

---

4.三种情况下运行结果的对比

在同一组测试数据下，三种情况的运行结果见上图。左为读者优先，中为公平策略，右为写者优先。可见左图读者进行了大量的插队操作，中图的读者与写者都是交替进行的，右图的写者从一开始就在插队。