手写阻塞队列（Condition实现）

Posted 2021-01-09 keeya

自己实现阻塞队列的话可以采用Object下的wait和notify方法，也可以使用Lock锁提供的Condition来实现，本文就是自己手撸的一个简单的阻塞队列，部分借鉴了JDK的源码。Ps：最近看面经的时候发现字节跳动的面试官特别喜欢让面试者手写阻塞队列，希望本文能对大家有帮助。个人手撸如有错误还请批评指正。

public class AxinBlockQueue {
    //队列容器
    private List<Integer> container = new ArrayList<>();
    private volatile int size;
    private volatile int capacity;
    private Lock lock = new ReentrantLock();
    //Condition
    private final Condition isNull = lock.newCondition();
    private final Condition isFull = lock.newCondition();

    AxinBlockQueue(int cap) {
        this.capacity = cap;
    }

    /**
     * 添加方法
     *
     * @param data
     */
    public void add(int data) {
        try {
            lock.lock();
            try {
                while (size >= capacity) {
                    System.out.println("阻塞队列满了");
                    isFull.await();
                }
            } catch (InterruptedException e) {
                isFull.signal();
                e.printStackTrace();
            }
            ++size;
            container.add(data);
            isNull.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 取出元素
     *
     * @return
     */
    public int take() {
        try {

            lock.lock();
            try {
                while (size == 0) {
                    System.out.println("阻塞队列空了");
                    isNull.await();
                }
            } catch (InterruptedException e) {
                isNull.signal();
                e.printStackTrace();
            }
            --size;
            int res = container.get(0);
            container.remove(0);
            isFull.signal();
            return res;
        } finally {
            lock.unlock();
        }
    }
}

下面是测试部分：更改生产者和消费者的延时可以看到阻塞队列满了和空了的效果。

public static void main(String[] args) {
    AxinBlockQueue queue = new AxinBlockQueue(5);
    Thread t1 = new Thread(() -> {
        for (int i = 0; i < 100; i++) {
            queue.add(i);
            System.out.println("塞入" + i);
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
    Thread t2 = new Thread(() -> {
        for (; ; ) {
            System.out.println("消费"+queue.take());
            try {
                Thread.sleep(800);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    });
    t1.start();
    t2.start();
}