Java多线程中的阻塞队列

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
public class TestBlockingQueue {

    //基于数组的方式实现一个普通队列
    //在改进成阻塞对列
    static class BlockingQueue{
        private int [] items = new int[10];
        //从head取元素
        private int head = 0;
        //从tail添加元素
        private int tail = 0;
        //表示队列中的元素
        private int size = 0;
        //引入一个锁对象
        private Object locker = new Object();

        //入队列
        public void put(int value) throws InterruptedException {
            synchronized (locker){
                //队列满后，阻塞等待，等到队列不满的时候，继续插入
                while(size == items.length){
                    locker.wait();
                }
                items[tail] = value;
                tail++;
                if(tail >= items.length){
                    tail = 0;
                }
                size++;
                locker.notifyAll();
            }
        }

        //出队列
        public Integer take() throws InterruptedException {
            int ret = 0;
            synchronized (locker){
            //队列为空同样进入阻塞等待，直到有元素时，继续出元素
                while(size == 0){
                    locker.wait();
                }
                 ret = items[head];
                head++;
                if(head >= items.length){
                    head = 0;
                }
                size --;
                locker.notifyAll();
            }
            return ret;
        }
    }
}

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
public static void main(String[] args) throws InterruptedException {
        BlockingQueue queue = new BlockingQueue();

        Thread producer = new Thread(){
            @Override
            public void run() {
                for (int i = 0; i < 30; i++) {
                    try {
                        System.out.println("生产了元素 " + i);
                        queue.put(i);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };

        Thread customer = new Thread(){
            @Override
            public void run() {
                while(true){
                    try {
                        int ret = queue.take();
                        System.out.println("消费了 " + ret);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };

        producer.start();
        customer.start();
    }