阻塞队列之ArrayBlockingQueue

痛定思痛。 2022-07-20 12:23 218阅读 0赞

### 阻塞队列 ###

> 当使用阻塞队列的时候，它可能会对当前线程产生阻塞，，比如一个线程从一个空的阻塞队列中取元素，此时线程会被阻塞直到阻塞队列中有了元素。当队列中有元素后，被阻塞的线程会自动被唤醒。

### 部分实现 ###

1.  ArrayBlockingQueue是使用数组作为底层存储的。

/** 数组存储 */
        private final E[] items;
        /** 队尾下标 */
        private int takeIndex;
        /** 队列头下标 */
        private int putIndex;
        /** 数组的长度 */
        private int count;

1.  offer()方法，将元素插入到队列，如果队列满了就返回false

public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            final ReentrantLock lock = this.lock;
            //如果没有被其它线程持有，便获取。否则等待
            lock.lock();
            try {
                if (count == items.length)
                    return false;
                else {
                    insert(e);
                    return true;
                }
            } finally {
                lock.unlock();
            }
        }

1.  put()方法，将指定的元素插入此队列的尾部，如果该队列已满，则等待可用的空间。

public void put(E e) throws InterruptedException {
            if (e == null) throw new NullPointerException();
            final E[] items = this.items;
            final ReentrantLock lock = this.lock;
            /**线程在请求lock并被阻塞时，如果被interrupt，则此线程会被唤醒并被要求处理InterruptedException*/
            lock.lockInterruptibly();
            try {
                try {
                    while (count == items.length)
                        notFull.await();
                } catch (InterruptedException ie) {
                    //唤醒一个等待线程。
                    notFull.signal();
                    throw ie;
                }
                insert(e);
            } finally {
                lock.unlock();
            }
        }

1.  poll()方法，获取并移除此队列的头，如果此队列为空，则返回 null。

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                if (count == 0)
                    return null;
                E x = extract();
                return x;
            } finally {
                lock.unlock();
            }
        }

1.  take()方法，获取并移除此队列的头部，在元素变得可用之前一直等待。

public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                try {
                    while (count == 0)
                        notEmpty.await();
                } catch (InterruptedException ie) {
                    notEmpty.signal(); // propagate to non-interrupted thread
                    throw ie;
                }
                E x = extract();
                return x;
            } finally {
                lock.unlock();
            }
        }

### 使用阻塞队列实现生产者、消费者 ###

public class BlockQueueTest { 
    
        ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(10);
    
        public static void main(String[] args) {
            BlockQueueTest test = new BlockQueueTest();
    
            Producer producer = test.new Producer();
            Consumer consumer = test.new Consumer();
    
            producer.start();
            consumer.start();
        }
    
        class Consumer extends Thread { 
    
            @Override
            public void run() {
                consume();
            }
    
            private void consume() {
                while (true) {
                    try {
                        queue.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        class Producer extends Thread { 
    
            @Override
            public void run() {
                produce();
            }
    
            private void produce() {
                while (true) {
                    try {
                        queue.put(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }