Blocking Queue Analysis

 

 ArrayBlockQueue put Method ： It's a lock , Judge whether the total number and length of the current array elements are the same , If the same block occurs , If it is less than the total number, it will be queued

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

 ArrayBlockQueue put And it's very simple ： Determine whether the total is 0, If 0 Then block .
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

 

 

 

 

 

 List

 

    //add by Hide on jdk  If the team is full or empty, it will directly report an error 
    new Thread(()->{
        try {
            TimeUnit.SECONDS.sleep(1);
            blockingQueue.add(1);
            blockingQueue.add(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"A").start();

    new Thread(()->{
        try {
            System.out.println(blockingQueue.remove());
            System.out.println("aaa");
            TimeUnit.SECONDS.sleep(1);
            System.out.println(blockingQueue.remove());
            System.out.println("bbbb");
            System.out.println(blockingQueue.remove());
            System.out.println("ccc");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"B").start();

    //add by Hide on jdk  Blocked when the team is full or empty 
    /*new Thread(()->{
        try {
            TimeUnit.SECONDS.sleep(1);
            blockingQueue.put(1);
            blockingQueue.put(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"A").start();

    new Thread(()->{
        try {
            System.out.println(blockingQueue.take());
            System.out.println("aaa");
            TimeUnit.SECONDS.sleep(1);
            System.out.println(blockingQueue.take());
            System.out.println("bbbb");
            System.out.println(blockingQueue.take());
            System.out.println("ccc");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    },"B").start();*/
}


/**
 * add  The method is to add an element to the queue , If the queue is full , An exception will be thrown to indicate that the queue is full .
 */
private static void addTest() {
    BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    System.out.println(blockingQueue.add(1));
    System.out.println(blockingQueue.add(2));
    System.out.println(blockingQueue.add(3));
}

/**
 * remove  The function of the method is to delete the element and return to the head node of the queue , If the deleted queue is empty , remove  Method will throw an exception .
 */
private static void removeTest() {
    ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    blockingQueue.add(1);
    blockingQueue.add(2);
    System.out.println(blockingQueue.remove());
    System.out.println(blockingQueue.remove());
    System.out.println(blockingQueue.remove());
}

/**
 * element  The method is to return the head node of the queue , But it doesn't delete . If the queue is empty , Throw an exception 
 */
private static void elementTest() {
    ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    blockingQueue.element();
}

/**
 * offer  Method is used to insert an element . If the addition is successful, it will return  true, And if the queue is full , return false
 */
private static void offerTest(){
    ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    System.out.println(blockingQueue.offer(1));
    System.out.println(blockingQueue.offer(2));
    System.out.println(blockingQueue.offer(3));
}

/**
 * poll  Method is also used to remove and return the head node of the queue .  If the queue is empty , return null
 */
private static void pollTest() {
    ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(3);
    blockingQueue.offer(1);
    blockingQueue.offer(2);
    blockingQueue.offer(3);
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
    System.out.println(blockingQueue.poll());
}

/**
 * peek  Method returns the header element of the queue, but does not delete .  If the queue is empty , return null
 */
private static void peekTest() {
    ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    System.out.println(blockingQueue.peek());
}

/**
 * put  The function of the method is to insert elements . If the queue is full, you cannot continue inserting , Blocking the insert thread , Until the queue is empty 
 */
private static void putTest(){
    BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(2);
    try {
        blockingQueue.put(1);
        blockingQueue.put(2);
        blockingQueue.put(3);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

/**
 * take  Method is used to get and remove the head node of the queue . If there is no data in the execution queue , The block , Until there's data in the queue 
 */
private static void takeTest(){
    try {
        blockingQueue.put(1);
        blockingQueue.put(2);
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println("bbb");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

LinkBlockingQueue: 

offer Core source code ： Less than the total capacity, directly enter the queue

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}