架构师养成记--5.线程之间的通信

1 import java.util.LinkedList;
2 import java.util.concurrent.TimeUnit;
3 import java.util.concurrent.atomic.AtomicInteger;
4
5 public class MyQueue {
6
7     //1 需要一个承装元素的集合
8     private LinkedList<Object> list = new LinkedList<Object>();
9
10     //2 需要一个计数器
11     private AtomicInteger count = new AtomicInteger(0);
12
13     //3 需要制定上限和下限
14     private final int minSize = 0;
15
16     private final int maxSize ;
17
18     //4 构造方法
19     public MyQueue(int size){
20         this.maxSize = size;
21     }
22
23     //5 初始化一个对象 用于加锁
24     private final Object lock = new Object();
25
26
27     //put(anObject): 把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断，直到BlockingQueue里面有空间再继续.
28     public void put(Object obj){
29         synchronized (lock) {
30             while(count.get() == this.maxSize){
31                 try {
32                     lock.wait();
33                 } catch (InterruptedException e) {
34                     e.printStackTrace();
35                 }
36             }
37             //1 加入元素
38             list.add(obj);
39             //2 计数器累加
40             count.incrementAndGet();
41             //3 通知另外一个线程（唤醒）
42             lock.notify();
43             System.out.println("新加入的元素为:" + obj);
44         }
45     }
46
47
48     //take: 取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入.
49     public Object take(){
50         Object ret = null;
51         synchronized (lock) {
52             while(count.get() == this.minSize){
53                 try {
54                     lock.wait();
55                 } catch (InterruptedException e) {
56                     e.printStackTrace();
57                 }
58             }
59             //1 做移除元素操作
60             ret = list.removeFirst();
61             //2 计数器递减
62             count.decrementAndGet();
63             //3 唤醒另外一个线程
64             lock.notify();
65         }
66         return ret;
67     }
68
69     public int getSize(){
70         return this.count.get();
71     }
72
73
74     public static void main(String[] args) {
75
76         final MyQueue mq = new MyQueue(5);
77         mq.put("a");
78         mq.put("b");
79         mq.put("c");
80         mq.put("d");
81         mq.put("e");
82
83         System.out.println("当前容器的长度:" + mq.getSize());
84
85         Thread t1 = new Thread(new Runnable() {
86             @Override
87             public void run() {
88                 mq.put("f");
89                 mq.put("g");
90             }
91         },"t1");
92
93         t1.start();
94
95
96         Thread t2 = new Thread(new Runnable() {
97             @Override
98             public void run() {
99                 Object o1 = mq.take();
100                 System.out.println("移除的元素为:" + o1);
101                 Object o2 = mq.take();
102                 System.out.println("移除的元素为:" + o2);
103             }
104         },"t2");
105
106
107         try {
108             TimeUnit.SECONDS.sleep(2);
109         } catch (InterruptedException e) {
110             e.printStackTrace();
111         }
112
113         t2.start();
114
115
116     }
117
118
119
120 }