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

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

1.读者优先

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

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

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

  1 package 读者优先;
  2 import java.util.Scanner;
  3 
  4 public class Main {
  5 
  6     public static void main(String[] args) {
  7         System.out.print("请设置读者数目:");
  8         Scanner scan=new Scanner(System.in);
  9         int readNum =scan.nextInt();
 10         
 11         System.out.print("请设置写者数目:");
 12         scan=new Scanner(System.in);
 13         int writeNum =scan.nextInt();        
 14         
 15         System.out.print("请设置循环上限:");
 16         scan=new Scanner(System.in);
 17         Global.UpBound =scan.nextInt();
 18         
 19         scan.close();
 20 
 21         
 22         Reader r[]=new Reader[readNum];
 23         Writer w[]=new Writer[writeNum];
 24         int i;
 25         for(i=0;i<readNum;i++){
 26             r[i]=new Reader(i+1);
 27         }
 28         for(i=0;i<writeNum;i++){
 29             w[i]=new Writer(i+1);
 30         }
 31         Thread []r_t=new Thread[readNum];
 32         Thread []w_t=new Thread[writeNum];
 33         for(i=0;i<readNum;i++){
 34             r_t[i]=new Thread(r[i]);
 35         }        
 36         for(i=0;i<writeNum;i++){
 37             w_t[i]=new Thread(w[i]);
 38         }
 39         for(i=0;i<writeNum;i++){
 40             w_t[i].start();
 41         }    
 42         for(i=0;i<readNum;i++){
 43             r_t[i].start();
 44         }    
 45         
 46 
 47     
 48     }
 49 
 50 }
 51 
 52 class syn{//PV操作类
 53     int count=0;//信号量
 54     syn(){}
 55     syn(int a){count=a;}
 56     public synchronized void Wait(){ //关键字 synchronized 保证了此操作是一条【原语】
 57         count--;
 58         if(count<0){//等于0 ：有一个进程进入了临界区
 59             try {         //小于0：abs(count)=阻塞的进程数目
 60                 this.wait();
 61             } catch (InterruptedException e) {
 62                 e.printStackTrace();  
 63             }  
 64         }  
 65     }  
 66     public synchronized void Signal(){   //关键字 synchronized 保证了此操作是一条【原语】
 67         count++;
 68         if(count<=0){//如果有进程阻塞
 69             this.notify();//All
 70         }
 71     }  
 72 }
 73 
 74 class Global{
 75     static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
 76     static syn mutex=new syn(1);//多个【写者】对数据区进行【互斥】访问
 77     static int dataZone=0;    //数据区
 78     static int readcount=0;    //用于记录读者的数量
 79     static int data=0;
 80     static int UpBound=20;
 81 }
 82 
 83 class Reader implements Runnable{//读者
 84     int ID=0;
 85     Reader(){}
 86     Reader(int id){ID=id;}
 87     public void run(){
 88         while(Global.data<=Global.UpBound){
 89             //对readcount进行操作
 90             Global.rmutex.Wait();
 91             if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
 92                 Global.mutex.Wait();
 93             }
 94             Global.readcount++;//读者数量增减
 95             Global.rmutex.Signal();
 96             //对readcount操作结束
 97             
 98             /*
 99              * 进行读操作
100              */
101             int readData=Global.dataZone;
102             System.out.println("读者"+ID+"读出了数据:"+readData);
103             try {
104                 Thread.sleep(100);
105             } catch (InterruptedException e) {
106                 e.printStackTrace();
107             }
108             /*
109              * 结束读操作
110              */
111             
112             //对readcount进行操作
113             Global.rmutex.Wait();
114             Global.readcount--;//读者数量减少
115             if(Global.readcount==0){//这是最后一个读者，唤醒写者
116                 Global.mutex.Signal();
117             }
118             Global.rmutex.Signal();
119             //对readcount操作结束
120         }
121     }
122 }
123 
124 class Writer implements Runnable{//写者
125     int ID=0;
126     Writer(){}
127     Writer(int id){ID=id;}
128     public void run(){
129         while(Global.data<=Global.UpBound){
130             Global.mutex.Wait();    //申请对数据区进行访问
131             /*
132              * 进行写操作
133              */
134             Global.data++;
135             int writeData=Global.data;
136             System.out.println("写者"+ID+"写入了数据:"+writeData);
137             Global.dataZone=Global.data;
138             try {
139                 Thread.sleep(10);
140             } catch (InterruptedException e) {
141                 // TODO Auto-generated catch block
142                 e.printStackTrace();
143             }
144             /*
145              * 结束写操作
146              */
147             Global.mutex.Signal();    //释放数据区，允许其他进程读写
148         }
149     }
150 }

View Code

2.公平策略

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

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

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

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

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

  1 package 公平策略;
  2 
  3 import java.util.Scanner;
  4 
  5 
  6 
  7 public class Main {
  8 
  9     public static void main(String[] args) {
 10         System.out.print("请设置读者数目:");
 11         Scanner scan=new Scanner(System.in);
 12         int readNum =scan.nextInt();
 13         
 14         System.out.print("请设置写者数目:");
 15         scan=new Scanner(System.in);
 16         int writeNum =scan.nextInt();        
 17         
 18         System.out.print("请设置循环上限:");
 19         scan=new Scanner(System.in);
 20         Global.UpBound =scan.nextInt();
 21         
 22         scan.close();
 23 
 24         
 25         Reader r[]=new Reader[readNum];
 26         Writer w[]=new Writer[writeNum];
 27         int i;
 28         for(i=0;i<readNum;i++){
 29             r[i]=new Reader(i+1);
 30         }
 31         for(i=0;i<writeNum;i++){
 32             w[i]=new Writer(i+1);
 33         }
 34         Thread []r_t=new Thread[readNum];
 35         Thread []w_t=new Thread[writeNum];
 36         for(i=0;i<readNum;i++){
 37             r_t[i]=new Thread(r[i]);
 38         }        
 39         for(i=0;i<writeNum;i++){
 40             w_t[i]=new Thread(w[i]);
 41         }
 42         for(i=0;i<writeNum;i++){
 43             w_t[i].start();
 44         }    
 45         for(i=0;i<readNum;i++){
 46             r_t[i].start();
 47         }    
 48         
 49 
 50     
 51     }
 52 
 53 }
 54 
 55 class syn{//PV操作类
 56     int count=0;//信号量
 57     syn(){}
 58     syn(int a){count=a;}
 59     public synchronized void Wait(){ //关键字 synchronized 保证了此操作是一条【原语】
 60         count--;
 61         if(count<0){//等于0 ：有一个进程进入了临界区
 62             try {         //小于0：abs(count)=阻塞的进程数目
 63                 this.wait();
 64             } catch (InterruptedException e) {
 65                 e.printStackTrace();  
 66             }  
 67         }  
 68     }  
 69     public synchronized void Signal(){   //关键字 synchronized 保证了此操作是一条【原语】
 70         count++;
 71         if(count<=0){//如果有进程阻塞
 72             this.notify();//All
 73         }
 74     }  
 75 }
 76 
 77 class Global{
 78     static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
 79     static syn mutex=new syn(1);//多个【写者】对数据区进行【互斥】访问
 80     static int dataZone=0;    //数据区
 81     static int readcount=0;    //用于记录读者的数量
 82     static int data=0;
 83     static int UpBound=20;
 84     static syn wmutex=new syn(1);//表示是否存在【进行】或【等待】的【写者】
 85 }
 86 
 87 class Reader implements Runnable{//读者
 88     int ID=0;
 89     Reader(){}
 90     Reader(int id){ID=id;}
 91     public void run(){
 92         while(Global.data<=Global.UpBound){
 93             Global.wmutex.Wait();//检测是否存在写者，无写者才能进入
 94             
 95             //对readcount进行操作
 96             Global.rmutex.Wait();
 97             if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
 98                 Global.mutex.Wait();
 99             }
100             Global.readcount++;//读者数量增减
101             Global.rmutex.Signal();
102             //对readcount操作结束
103             
104             Global.wmutex.Signal();//恢复wmutex
105             
106             /*
107              * 进行读操作
108              */
109             int readData=Global.dataZone;
110             System.out.println("读者"+ID+"读出了数据:"+readData);
111             try {
112                 Thread.sleep(100);
113             } catch (InterruptedException e) {
114                 e.printStackTrace();
115             }
116             /*
117              * 结束读操作
118              */
119             
120             //对readcount进行操作
121             Global.rmutex.Wait();
122             Global.readcount--;//读者数量减少
123             if(Global.readcount==0){//这是最后一个读者，唤醒写者
124                 Global.mutex.Signal();
125             }
126             Global.rmutex.Signal();
127             //对readcount操作结束
128             
129             
130         }
131     }
132 }
133 
134 class Writer implements Runnable{//写者
135     int ID=0;
136     Writer(){}
137     Writer(int id){ID=id;}
138     public void run(){
139         while(Global.data<=Global.UpBound){
140             Global.wmutex.Wait();
141             Global.mutex.Wait();    //申请对数据区进行访问
142             /*
143              * 进行写操作
144              */
145             Global.data++;
146             int writeData=Global.data;
147             System.out.println("写者"+ID+"写入了数据:"+writeData);
148             Global.dataZone=Global.data;
149             try {
150                 Thread.sleep(10);
151             } catch (InterruptedException e) {
152                 // TODO Auto-generated catch block
153                 e.printStackTrace();
154             }
155             /*
156              * 结束写操作
157              */
158             Global.mutex.Signal();    //释放数据区，允许其他进程读写
159             Global.wmutex.Signal();
160         }
161     }
162 }

View Code

3.写者优先

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

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

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

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

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

  1 package 写者优先;
  2 
  3 import java.util.Scanner;
  4 
  5 
  6 
  7 public class Main {
  8 
  9     public static void main(String[] args) {
 10         System.out.print("请设置读者数目:");
 11         Scanner scan=new Scanner(System.in);
 12         int readNum =scan.nextInt();
 13         
 14         System.out.print("请设置写者数目:");
 15         scan=new Scanner(System.in);
 16         int writeNum =scan.nextInt();        
 17         
 18         System.out.print("请设置循环上限:");
 19         scan=new Scanner(System.in);
 20         Global.UpBound =scan.nextInt();
 21         
 22         scan.close();
 23 
 24         
 25         Reader r[]=new Reader[readNum];
 26         Writer w[]=new Writer[writeNum];
 27         int i;
 28         for(i=0;i<readNum;i++){
 29             r[i]=new Reader(i+1);
 30         }
 31         for(i=0;i<writeNum;i++){
 32             w[i]=new Writer(i+1);
 33         }
 34         Thread []r_t=new Thread[readNum];
 35         Thread []w_t=new Thread[writeNum];
 36         for(i=0;i<readNum;i++){
 37             r_t[i]=new Thread(r[i]);
 38         }        
 39         for(i=0;i<writeNum;i++){
 40             w_t[i]=new Thread(w[i]);
 41         }
 42         for(i=0;i<writeNum;i++){
 43             w_t[i].start();
 44         }    
 45         for(i=0;i<readNum;i++){
 46             r_t[i].start();
 47         }    
 48         
 49 
 50     
 51     }
 52 
 53 }
 54 
 55 class syn{//PV操作类
 56     int count=0;//信号量
 57     syn(){}
 58     syn(int a){count=a;}
 59     public synchronized void Wait(){ //关键字 synchronized 保证了此操作是一条【原语】
 60         count--;
 61         if(count<0){//等于0 ：有一个进程进入了临界区
 62             try {         //小于0：abs(count)=阻塞的进程数目
 63                 this.wait();
 64             } catch (InterruptedException e) {
 65                 e.printStackTrace();  
 66             }  
 67         }  
 68     }  
 69     public synchronized void Signal(){   //关键字 synchronized 保证了此操作是一条【原语】
 70         count++;
 71         if(count<=0){//如果有进程阻塞
 72             this.notify();//All
 73         }
 74     }  
 75 }
 76 
 77 class Global{
 78     static syn mutex=new syn(1);//控制互斥访问的数据区    
 79     static syn rmutex=new syn(1);//多个【读者】对readcount进行【互斥】访问
 80     static syn wmutex=new syn(1);//多个【写者】对writecount进行【互斥】访问
 81     static syn readable=new syn(1);//表示当前是否有写者
 82     
 83     static int dataZone=0;    //数据区
 84     static int readcount=0;    //用于记录读者的数量
 85     static int writecount=0;    //用于记录读者的数量
 86     
 87     static int data=0;
 88     static int UpBound=20;
 89 
 90 }
 91 
 92 class Reader implements Runnable{//读者
 93     int ID=0;
 94     Reader(){}
 95     Reader(int id){ID=id;}
 96     public void run(){
 97         while(Global.data<=Global.UpBound){
 98             Global.readable.Wait();//检测是否存在写者，无写者才能进入
 99             
100             //对readcount进行操作
101             Global.rmutex.Wait();
102             if(Global.readcount==0){//这是第一个读者，应该阻止写者的进入
103                 Global.mutex.Wait();
104             }
105             Global.readcount++;//读者数量增减
106             Global.rmutex.Signal();
107             //对readcount操作结束
108             
109             Global.readable.Signal();//恢复readable
110             
111             /*
112              * 进行读操作
113              */
114             int readData=Global.dataZone;
115             System.out.println("读者"+ID+"读出了数据:"+readData);
116             try {
117                 Thread.sleep(100);
118             } catch (InterruptedException e) {
119                 e.printStackTrace();
120             }
121             /*
122              * 结束读操作
123              */
124             
125             //对readcount进行操作
126             Global.rmutex.Wait();
127             Global.readcount--;//读者数量减少
128             if(Global.readcount==0){//这是最后一个读者，唤醒写者
129                 Global.mutex.Signal();
130             }
131             Global.rmutex.Signal();
132             //对readcount操作结束
133         }
134     }
135 }
136 
137 class Writer implements Runnable{//写者
138     int ID=0;
139     Writer(){}
140     Writer(int id){ID=id;}
141     public void run(){
142         while(Global.data<=Global.UpBound){
143             Global.wmutex.Wait();//准备修改writecount
144             if(Global.writecount==0) Global.readable.Wait();//如果是第一个读者，则阻止后续读者进入
145             Global.writecount++;
146             Global.wmutex.Signal();//结束对writecount的修改
147             
148             Global.mutex.Wait();    //申请对数据区进行访问
149             /*
150              * 进行写操作
151              */
152             Global.data++;
153             int writeData=Global.data;
154             System.out.println("写者"+ID+"写入了数据:"+writeData);
155             Global.dataZone=Global.data;
156             try {
157                 Thread.sleep(10);
158             } catch (InterruptedException e) {
159                 // TODO Auto-generated catch block
160                 e.printStackTrace();
161             }
162             /*
163              * 结束写操作
164              */
165             Global.mutex.Signal();    //释放数据区，允许其他进程读写
166 
167             Global.wmutex.Wait();//准备修改writecount
168             Global.writecount--;            
169             if(Global.writecount==0) Global.readable.Signal();//如果是最后一个写者，唤醒读者
170             Global.wmutex.Signal();//结束对writecount的修改
171         }
172     }
173 }

View Code

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

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