java实现的多生产者多消费者模型

用LinkedList实现的生产者消费者模型。
注意：对LinkedList的大小没有作限制，但在实际应用过程中该队列的大小应该是要做限制，以免耗尽内存。当对LinkedList的最大元素个数做限制后，实际上就成了一个又固定大小的循环队列。此时要对生产者的行为做一个约定，是覆盖老的数据，还是等待消费者消费掉数据后，又空位置再继续生产。

这里的程序没有对LinkedList大小做限制，代码如下：

点击(此处)折叠或打开

1. package ProCon;
   
2. 
   
3. import java.util.LinkedList;
   
4. import java.util.Random;
   
5. import java.util.concurrent.SynchronousQueue;
   
6. 
   
7. import javax.naming.InterruptedNamingException;
   
8. 
   
9. class Worker {
   
10.     LinkedList<String> JobQueue = new LinkedList<String>();
    
11.     
    
12.     Worker() {}
    
13.     
    
14.     // consumer
    
15.     public class Consumer implements Runnable {
    
16.         String item = null;
    
17.         
    
18.         public void run() {
    
19.             for ( ; ; ) {
    
20.                 synchronized (JobQueue) {
    
21.                     System.out.println("cons get lock");
    
22.                     while (JobQueue.size() == 0) {
    
23.                         System.out.println("cons wait . release lock");
    
24.                             try {
    
25.                                 JobQueue.wait();
    
26.                             } catch (InterruptedException e) {
    
27.                                 // TODO Auto-generated catch block
    
28.                                 e.printStackTrace();
    
29.                             }
    
30.                     }
    
31.                     item = JobQueue.remove();
    
32.                     System.out.println("consumer get: " + item);
    
33.                 }
    
34.                 
    
35.                 // wait for a while
    
36.                 try {
    
37.                     Thread.sleep(1000);
    
38.                 } catch (InterruptedException e) {
    
39.                     // TODO Auto-generated catch block
    
40.                     e.printStackTrace();
    
41.                 }
    
42.             }
    
43.             
    
44.         }
    
45.     }
    
46.     
    
47.     // producer
    
48.     public class Producer implements Runnable {
    
49.         Random r;
    
50.         StringBuffer sb = new StringBuffer();
    
51.         
    
52.         Producer() {
    
53.             r = new Random();
    
54.         }
    
55.         public void run() {
    
56.             for ( ; ; ) {
    
57.                 System.out.println("producer loop ");
    
58.                 synchronized (JobQueue) {
    
59.                     System.out.println("producer get lock");
    
60.                     sb.delete(0, sb.length());
    
61.                     sb.append("").append(r.nextInt(100));
    
62.                     System.out.println("put : " + sb.toString());
    
63. 
    
64.                     JobQueue.add(sb.toString());
    
65.                     JobQueue.notifyAll();
    
66.                 }
    
67.                 
    
68.                 System.out.println("producer release lock");
    
69.                 // wait for a while
    
70.                 try {
    
71.                     Thread.sleep(1000);
    
72.                 } catch (InterruptedException e) {
    
73.                     // TODO Auto-generated catch block
    
74.                     e.printStackTrace();
    
75.                 }
    
76.             }
    
77.         }
    
78.     }
    
79.     
    
80.     
    
81.     // work produce
    
82.     public void run() {
    
83.         Producer p = new Producer();
    
84.         Consumer c = new Consumer();
    
85.         
    
86.         new Thread(p).start();
    
87.         new Thread(p).start();
    
88.         
    
89.         // consumer
    
90.         new Thread(c).start();
    
91.         new Thread(c).start();
    
92.         new Thread(c).start();
    
93.     }
    
94. }
    
95. 
    
96. public class TestProCon {
    
97.     public static void main(String[] args) throws InterruptedException {
    
98.         Worker w = new Worker();
    
99.         
    
100.         w.run();
     
101.         System.out.println("end main");
     
102.         Thread.currentThread().join();
     
103.     }
     
104. }

运行效果如下：

点击(此处)折叠或打开

1. producer loop
2. producer get lock
3. Thread-0 put : 56
4. producer release lock
5. producer loop
6. producer get lock
7. Thread-1 put : 36
8. producer release lock
9. cons get lock
10. end main
11. Thread-2 consumer get: 56
12. cons get lock
13. Thread-3 consumer get: 36
14. cons get lock
15. cons wait . release lock
16. producer loop
17. producer get lock
18. Thread-0 put : 98
19. Thread-4 consumer get: 98
20. producer release lock
21. producer loop
22. producer get lock
23. Thread-1 put : 36
24. producer release lock
25. cons get lock
26. Thread-2 consumer get: 36
27. cons get lock
28. cons wait . release lock
29. cons get lock
30. cons wait . release lock
31. producer loop
32. producer get lock
33. Thread-0 put : 15
34. producer release lock
35. Thread-4 consumer get: 15
36. cons wait . release lock
37. producer loop
38. cons get lock
39. cons wait . release lock
40. producer get lock
41. Thread-1 put : 30
42. producer release lock
43. Thread-2 consumer get: 30
44. cons wait . release lock
45. producer loop
46. cons get lock
47. cons wait . release lock
48. producer get lock
49. Thread-0 put : 79
50. producer release lock
51. Thread-4 consumer get: 79
52. cons wait . release lock
53. producer loop
54. cons get lock
55. cons wait . release lock
56. producer get lock
57. Thread-1 put : 13
58. producer release lock
59. Thread-2 consumer get: 13
60. cons wait . release lock
61. producer loop
62. cons get lock
63. cons wait . release lock
64. producer get lock
65. Thread-0 put : 70
66. Thread-4 consumer get: 70
67. producer release lock
68. cons wait . release lock
69. cons get lock
70. cons wait . release lock
71. producer loop
72. producer get lock
73. Thread-1 put : 15
74. producer release lock
75. Thread-2 consumer get: 15
76. cons wait . release lock

从结果中可以看到是多个生产者，和多个消费者同时在运行。