windows下生产者消费者实现[c++ & java]

点击(此处)折叠或打开

1. // 生产者消费者.cpp : 定义控制台应用程序的入口点。
   
2. //
   
3. 
   
4. #include "stdafx.h"
   
5. #include <iostream>
   
6. #include <Windows.h>
   
7. using namespace std;
   
8. 
   
9. const int BUFFER_SIZE=5;
   
10. const int producerCount=5;
    
11. const int consumerCount=5;
    
12. const int timewait=INFINITE;
    
13. static int pID=0;
    
14. int cID=0;
    
15. HANDLE mutex;
    
16. HANDLE empty;
    
17. HANDLE full;
    
18. int in=0,out=0;
    
19. int buffer[BUFFER_SIZE];
    
20. 
    
21. DWORD WINAPI pFunc(LPVOID lparam);
    
22. DWORD WINAPI cFunc(LPVOID lparam);
    
23. 
    
24. int _tmain(int argc, _TCHAR* argv[])
    
25. {
    
26.     HANDLE pThread[producerCount];
    
27.     HANDLE cThread[consumerCount];
    
28.     DWORD producerID[producerCount];
    
29.     DWORD consumerID[consumerCount];
    
30. 
    
31.     mutex=CreateMutex(NULL,FALSE,NULL);
    
32.     empty=CreateSemaphore(NULL,BUFFER_SIZE,BUFFER_SIZE,NULL);
    
33.     full =CreateSemaphore(NULL,0,BUFFER_SIZE,NULL);
    
34. 
    
35.     for(int i=0 ; i<producerCount ; ++i){
    
36.         pThread[i]=CreateThread(NULL,0,pFunc,NULL,0,&producerID[i]);
    
37.         if(pThread[i]==NULL)
    
38.             return -1;
    
39.     }
    
40.     
    
41.     for(int i=0 ; i<consumerCount ; ++i){
    
42.         cThread[i]=CreateThread(NULL,0,cFunc,NULL,0,&consumerID[i]);
    
43.         if(cThread[i]==NULL)
    
44.             return -1;
    
45.     }
    
46.     
    
47.     bool isContinue=true;
    
48.     while(isContinue)
    
49.     {
    
50.         if(getchar())
    
51.         {
    
52.             isContinue = false;
    
53.         }
    
54.     }
    
55. 
    
56.     for(int i=0 ; i<producerCount ; ++i){
    
57.         CloseHandle(pThread[i]);
    
58.     }
    
59.     for(int i=0 ; i<consumerCount ; ++i){
    
60.         CloseHandle(cThread[i]);
    
61.     }
    
62.     CloseHandle(mutex);
    
63.     CloseHandle(empty);
    
64.     CloseHandle(full);
    
65.     return 0;
    
66. }
    
67. 
    
68. DWORD WINAPI pFunc(LPVOID lparam)
    
69. {
    
70.     Sleep(100);
    
71. 
    
72.     WaitForSingleObject(empty,timewait);
    
73.     WaitForSingleObject(mutex,timewait);
    
74. 
    
75.     cout<<"produced one ID:"<<pID<<endl;
    
76.     buffer[in]=pID;
    
77.     in=(in+1)%BUFFER_SIZE;
    
78.     pID++;
    
79. 
    
80.     ReleaseMutex(mutex);
    
81.     ReleaseSemaphore(full,1,NULL);
    
82. 
    
83.     return 0;
    
84. }
    
85. 
    
86. DWORD WINAPI cFunc(LPVOID lparam)
    
87. {
    
88.     Sleep(100);
    
89. 
    
90.     WaitForSingleObject(full,timewait);
    
91.     WaitForSingleObject(mutex,timewait);
    
92. 
    
93.     cID=buffer[out];
    
94.     out=(out+1)%BUFFER_SIZE;
    
95.     cout<<"consumed one ID:"<<cID<<endl;
    
96.     
    
97.     ReleaseMutex(mutex);
    
98.     ReleaseSemaphore(empty,1,NULL);
    
99. 
    
100.     return 0;
     
101. }

java实现：

点击(此处)折叠或打开

1. import java.util.Date;
   
2. import java.util.concurrent.Semaphore;
   
3. 
   
4. class Producer implements Runnable
   
5. {
   
6.     private BoundedBuffer buffer;
   
7.     private int producerCount=5;
   
8.     private long sleeptime=100;
   
9.     
   
10.     public Producer(BoundedBuffer buffer){
    
11.         this.buffer=buffer;
    
12.     }
    
13.     
    
14.     public void run(){
    
15.         Date message;
    
16.         for(int i=0 ;i<producerCount ;i++){
    
17.             try {
    
18.                 Thread.sleep(sleeptime);
    
19.             } catch (InterruptedException e) {
    
20.                 // TODO Auto-generated catch block
    
21.                 e.printStackTrace();
    
22.             }
    
23.             message=new Date();
    
24.             buffer.insert(message);
    
25.             System.out.println("in producer:"+i+" "+message);
    
26.         }
    
27.     }
    
28. }
    
29. 
    
30. class Consumer implements Runnable
    
31. {
    
32.     private BoundedBuffer buffer;
    
33.     private int consumerCount=5;
    
34.     private long sleeptime=100;
    
35.     
    
36.     public Consumer(BoundedBuffer buffer){
    
37.         this.buffer=buffer;
    
38.     }
    
39.     
    
40.     public void run(){
    
41.         Date message;
    
42.         for(int i=0 ;i<consumerCount ;i++){
    
43.             try {
    
44.                 Thread.sleep(sleeptime);
    
45.             } catch (InterruptedException e) {
    
46.                 // TODO Auto-generated catch block
    
47.                 e.printStackTrace();
    
48.             }
    
49.             message=(Date)buffer.remove();
    
50.             System.out.println("in consumer:"+i+" "+message);
    
51.         }
    
52.     }
    
53. }
    
54. 
    
55. class BoundedBuffer
    
56. {
    
57.     private static final int BUFFER_SIZE=5;
    
58.     private Object[] buffer;
    
59.     private int in,out;
    
60.     private Semaphore mutex;
    
61.     private Semaphore empty;
    
62.     private Semaphore full;
    
63.     
    
64.     public BoundedBuffer(){
    
65.         in=0;
    
66.         out=0;
    
67.         buffer=new Object[BUFFER_SIZE];
    
68.         
    
69.         mutex=new Semaphore(1);
    
70.         empty=new Semaphore(BUFFER_SIZE);
    
71.         full=new Semaphore(0);
    
72.     }
    
73.     
    
74.     public void insert(Object item){
    
75.         try {
    
76.             empty.acquire();
    
77.         } catch (InterruptedException e) {
    
78.             // TODO Auto-generated catch block
    
79.             e.printStackTrace();
    
80.         }
    
81.         try {
    
82.             mutex.acquire();
    
83.         } catch (InterruptedException e) {
    
84.             // TODO Auto-generated catch block
    
85.             e.printStackTrace();
    
86.         }
    
87.         
    
88.         buffer[in]=item;
    
89.         in=(in+1)%BUFFER_SIZE;
    
90.         
    
91.         mutex.release();
    
92.         full.release();
    
93.     }
    
94.     
    
95.     public Object remove(){
    
96.         try {
    
97.             full.acquire();
    
98.         } catch (InterruptedException e) {
    
99.             // TODO Auto-generated catch block
    
100.             e.printStackTrace();
     
101.         }
     
102.         try {
     
103.             mutex.acquire();
     
104.         } catch (InterruptedException e) {
     
105.             // TODO Auto-generated catch block
     
106.             e.printStackTrace();
     
107.         }
     
108.         
     
109.         Object item=buffer[out];
     
110.         out=(out+1)%BUFFER_SIZE;
     
111.         
     
112.         mutex.release();
     
113.         empty.release();
     
114.         return item;
     
115.     }
     
116. }
     
117. 
     
118. public class 生产者消费者
     
119. {
     
120.     public static void main(String[] args){
     
121.         BoundedBuffer buffer=new BoundedBuffer();
     
122.         
     
123.         Thread producer=new Thread(new Producer(buffer));
     
124.         Thread consumer=new Thread(new Consumer(buffer));
     
125.         
     
126.         producer.start();
     
127.         consumer.start();
     
128.     }
     
129. }

菜鸟学习中，望路过大神多多指点。