ArrayBlockingQueue中几种添加和取出元素方法的解析直接通过效果演示是非常好的API学习方式。简单的源码分析

101 阅读 0 评论 67 点赞

我是靠谱客的博主端庄樱桃，这篇文章主要介绍ArrayBlockingQueue中几种添加和取出元素方法的解析直接通过效果演示是非常好的API学习方式。简单的源码分析，现在分享给大家，希望可以做个参考。

直接通过效果演示是非常好的API学习方式。

复制代码

public class TestBlockingQueue {
    public static void main(String[] args) {
        testAdd();

System.out.println("=====================我是分隔符=========================");

testOffer();

System.out.println("=====================我是分隔符=========================");

testPut();

System.out.println("=====================我是分隔符=========================");

testPoll();

System.out.println("=====================我是分隔符=========================");

testPeek();

System.out.println("=====================我是分隔符=========================");

testRemove();

System.out.println("=====================我是分隔符=========================");

testTake();

System.out.println("=====================我是分隔符=========================");

testClear();

}

private static void testClear() {

ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

System.out.println("当前队列中的元素为：" + abq.toString());

/**
         * 清空队列中的所有元素
         */
        abq.clear();

System.out.println("调用clear方法后，当前队列中的元素为：" + abq.toString());

}

private static void testTake() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

try {
            /**
             * 从队列头部获取元素（如果队列为空，则阻塞），返回取到的结果。
             */
            Integer i = abq.take();
            System.out.println("调用take方法，返回：" + i);

i = abq.take();
            System.out.println("调用take方法，返回：" + i);

new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("等待1秒后，我是一个新的线程，我向队列中放入一个元素！");
                    abq.put(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();

System.out.println("调用take方法阻塞了！");
            long s = System.currentTimeMillis();
            i = abq.take();
            long e = System.currentTimeMillis();

System.out.println("阻塞了" + (e - s) / 1000 + "秒！返回值：" + i);

} catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

private static void testRemove() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

System.out.println("当前队列中的元素为：" + abq.toString());

/**
         * 从队列头部移除一个元素，成功则返回被移除的元素，否则抛出NoSuchElementException异常，这是AbstractQueue父类的方法
         */
        Integer i = abq.remove();
        System.out.println("remove方法移除的元素为：" + i);

i = abq.remove();
        System.out.println("remove方法被移除的元素为：" + i);
        try {
            abq.remove();
        } catch (NoSuchElementException e) {
            System.out.println("队列为空，调用remove抛出NoSuchElementException异常");
        }

abq = new ArrayBlockingQueue<>(5);
        abq.add(1);
        abq.add(2);
        abq.add(3);
        abq.add(5);
        abq.add(1);

System.out.println("当前队列中的元素为：" + abq.toString());

/**
         * 从头部移除1个指定的元素，成功则返回true，否则返回false
         */

boolean remove = abq.remove(1);

System.out.println("移除元素1，返回：" + remove);

System.out.println("移除后队列中的元素为：" + abq.toString());

remove = abq.remove(6);

System.out.println("移除元素6，返回：" + remove);
    }

private static void testPeek() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

/**
         * 从队列头部获取元素（非阻塞，与poll方法不同点在于，只是获取元素，并不会把元素从队列中移除），如果取到则返回取到的结果，否则返回null
         */

Integer i = abq.peek();

System.out.println("调用peek方法，返回：" + i);

System.out.println("当前队列中的元素为：" + abq.toString());

}

private static void testPoll() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

/**
         * 从队列头部取出元素（非阻塞，也可以指定阻塞等待时间），如果取到则返回取到的结果，否则返回null
         */

Integer i = abq.poll();

System.out.println("调用poll方法，返回：" + i);

System.out.println("当前队列中的元素为：" + abq.toString());

i = abq.poll();

System.out.println("调用poll方法，返回：" + i);

System.out.println("当前队列中的元素为：" + abq.toString());

try {
            long s = System.currentTimeMillis();
            i = abq.poll(1, TimeUnit.SECONDS);
            long e = System.currentTimeMillis();
            System.out.println("等待了" + (e - s) / 1000 + "秒后，返回：" + i);

System.out.println("当前队列中的元素为：" + abq.toString());

} catch (InterruptedException e) {
            e.printStackTrace();
        }

}

private static void testPut() {

ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        try {
            /**
             * 向队列的尾部插入元素（如果队列满了则阻塞。）。
             */
            abq.put(1);
            System.out.println("当前队列中的元素为：" + abq.toString());
            abq.put(2);
            System.out.println("当前队列中的元素为：" + abq.toString());

new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("等待1秒后，我是一个新的线程，我从队列中取出一个元素！");
                    abq.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();

System.out.println("调用put方法阻塞了！");
            long s = System.currentTimeMillis();
            abq.put(3);
            long e = System.currentTimeMillis();

System.out.println("阻塞了" + (e - s) / 1000 + "秒！当前队列中的元素为" + abq.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

private static void testOffer() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        boolean offer = abq.offer(1);
        /**
         * 向队列的尾部插入元素（非阻塞，也可以指定阻塞等待时间），成功添加返回true，否则返回false
         */
        System.out.println("调用offer方法，当前返回：" + offer);

offer = abq.offer(2);

System.out.println("调用offer方法，当前返回：" + offer);

offer = abq.offer(3);

System.out.println("调用offer方法，队列已满，当前返回：" + offer);

try {
            long s = System.currentTimeMillis();
            offer = abq.offer(3, 1, TimeUnit.SECONDS);
            long e = System.currentTimeMillis();
            System.out.println("等待了" + (e - s) / 1000 + "秒后，返回：" + offer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

System.out.println("当前队列中的元素为：" + abq.toString());

}

private static void testAdd() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        boolean add = abq.add(1);
        /**
         * 向队列的尾部插入元素（非阻塞），成功添加返回true，否则会抛出IllegalStateException异常，这是AbstractQueue父类的方法
         */
        System.out.println("调用add方法，当前返回：" + add);
        try {
            abq.add(2);
            abq.add(3);
        } catch (IllegalStateException e) {
            System.out.println("调用add方法，队列已满，抛出IllegalStateException异常：" + e.getMessage());
        }
        System.out.println("当前队列中的元素为：" + abq.toString());
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
public class TestBlockingQueue {
    public static void main(String[] args) {
        testAdd();

        System.out.println("=====================我是分隔符=========================");

        testOffer();

        System.out.println("=====================我是分隔符=========================");

        testPut();

        System.out.println("=====================我是分隔符=========================");

        testPoll();

        System.out.println("=====================我是分隔符=========================");

        testPeek();

        System.out.println("=====================我是分隔符=========================");

        testRemove();

        System.out.println("=====================我是分隔符=========================");

        testTake();

        System.out.println("=====================我是分隔符=========================");

        testClear();

    }

    private static void testClear() {

        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

        System.out.println("当前队列中的元素为：" + abq.toString());

        /**
         * 清空队列中的所有元素
         */
        abq.clear();

        System.out.println("调用clear方法后，当前队列中的元素为：" + abq.toString());

    }

    private static void testTake() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

        try {
            /**
             * 从队列头部获取元素（如果队列为空，则阻塞），返回取到的结果。
             */
            Integer i = abq.take();
            System.out.println("调用take方法，返回：" + i);

            i = abq.take();
            System.out.println("调用take方法，返回：" + i);

            new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("等待1秒后，我是一个新的线程，我向队列中放入一个元素！");
                    abq.put(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();

            System.out.println("调用take方法阻塞了！");
            long s = System.currentTimeMillis();
            i = abq.take();
            long e = System.currentTimeMillis();

            System.out.println("阻塞了" + (e - s) / 1000 + "秒！返回值：" + i);

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void testRemove() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

        System.out.println("当前队列中的元素为：" + abq.toString());

        /**
         * 从队列头部移除一个元素，成功则返回被移除的元素，否则抛出NoSuchElementException异常，这是AbstractQueue父类的方法
         */
        Integer i = abq.remove();
        System.out.println("remove方法移除的元素为：" + i);

        i = abq.remove();
        System.out.println("remove方法被移除的元素为：" + i);
        try {
            abq.remove();
        } catch (NoSuchElementException e) {
            System.out.println("队列为空，调用remove抛出NoSuchElementException异常");
        }


        abq = new ArrayBlockingQueue<>(5);
        abq.add(1);
        abq.add(2);
        abq.add(3);
        abq.add(5);
        abq.add(1);

        System.out.println("当前队列中的元素为：" + abq.toString());

        /**
         * 从头部移除1个指定的元素，成功则返回true，否则返回false
         */

        boolean remove = abq.remove(1);

        System.out.println("移除元素1，返回：" + remove);

        System.out.println("移除后队列中的元素为：" + abq.toString());

        remove = abq.remove(6);

        System.out.println("移除元素6，返回：" + remove);
    }

    private static void testPeek() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

        /**
         * 从队列头部获取元素（非阻塞，与poll方法不同点在于，只是获取元素，并不会把元素从队列中移除），如果取到则返回取到的结果，否则返回null
         */

        Integer i = abq.peek();

        System.out.println("调用peek方法，返回：" + i);

        System.out.println("当前队列中的元素为：" + abq.toString());

    }

    private static void testPoll() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        abq.add(1);
        abq.add(2);

        /**
         * 从队列头部取出元素（非阻塞，也可以指定阻塞等待时间），如果取到则返回取到的结果，否则返回null
         */

        Integer i = abq.poll();

        System.out.println("调用poll方法，返回：" + i);

        System.out.println("当前队列中的元素为：" + abq.toString());

        i = abq.poll();

        System.out.println("调用poll方法，返回：" + i);

        System.out.println("当前队列中的元素为：" + abq.toString());

        try {
            long s = System.currentTimeMillis();
            i = abq.poll(1, TimeUnit.SECONDS);
            long e = System.currentTimeMillis();
            System.out.println("等待了" + (e - s) / 1000 + "秒后，返回：" + i);

            System.out.println("当前队列中的元素为：" + abq.toString());

        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private static void testPut() {

        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        try {
            /**
             * 向队列的尾部插入元素（如果队列满了则阻塞。）。
             */
            abq.put(1);
            System.out.println("当前队列中的元素为：" + abq.toString());
            abq.put(2);
            System.out.println("当前队列中的元素为：" + abq.toString());

            new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("等待1秒后，我是一个新的线程，我从队列中取出一个元素！");
                    abq.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();

            System.out.println("调用put方法阻塞了！");
            long s = System.currentTimeMillis();
            abq.put(3);
            long e = System.currentTimeMillis();

            System.out.println("阻塞了" + (e - s) / 1000 + "秒！当前队列中的元素为" + abq.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void testOffer() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        boolean offer = abq.offer(1);
        /**
         * 向队列的尾部插入元素（非阻塞，也可以指定阻塞等待时间），成功添加返回true，否则返回false
         */
        System.out.println("调用offer方法，当前返回：" + offer);

        offer = abq.offer(2);

        System.out.println("调用offer方法，当前返回：" + offer);

        offer = abq.offer(3);

        System.out.println("调用offer方法，队列已满，当前返回：" + offer);

        try {
            long s = System.currentTimeMillis();
            offer = abq.offer(3, 1, TimeUnit.SECONDS);
            long e = System.currentTimeMillis();
            System.out.println("等待了" + (e - s) / 1000 + "秒后，返回：" + offer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("当前队列中的元素为：" + abq.toString());

    }

    private static void testAdd() {
        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(2);
        boolean add = abq.add(1);
        /**
         * 向队列的尾部插入元素（非阻塞），成功添加返回true，否则会抛出IllegalStateException异常，这是AbstractQueue父类的方法
         */
        System.out.println("调用add方法，当前返回：" + add);
        try {
            abq.add(2);
            abq.add(3);
        } catch (IllegalStateException e) {
            System.out.println("调用add方法，队列已满，抛出IllegalStateException异常：" + e.getMessage());
        }
        System.out.println("当前队列中的元素为：" + abq.toString());
    }

}

简单的源码分析

ArrayBlockingQueue主要是通过lock和Condition来实现阻塞逻辑的。

复制代码

1
2
3
4
5
6
7
8
9
    /** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;

put

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lockInterruptibly();
        try {
            while (count == items.length)
            	//容器中的数量达到容器上限时
            	//阻塞，等待取出元素后调用notFull.signal()唤醒。
                notFull.await();
            //否则入队    
            enqueue(e);
        } finally {
        	//释放锁
            lock.unlock();
        }
    }

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        //下标从0开始放元素
        items[putIndex] = x;
        //如果容器满了，下标移回到0位置
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        //元素入队后，通知notEmpty，表示容器肯定不为空，可以获取。
        notEmpty.signal();
    }

take

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lockInterruptibly();
        try {
            while (count == 0)
            	//如果容器中没有元素了，则阻塞，等待添加完元素后调用notEmpty.signal()唤醒。
                notEmpty.await();
                //出队
            return dequeue();
        } finally {
        	//释放锁
            lock.unlock();
        }
    }

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        //从下标0开始出队，拿到出队的元素
        E x = (E) items[takeIndex];
        //删除引用
        items[takeIndex] = null;
        //下标循环
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
         //元素出队后，通知notFull，表示容器肯定不满了，可以继续存放。    
        notFull.signal();
        return x;
    }