Blocking queue

1. What is a blocking queue

A blocking queue is a special queue , Also abide by " fifo " Principles . Be careful not to contact with the operating system kernel , Indicates the blocking state PCB The linked list of is confused ！！

Blocking queues can ensure thread safety , It has the following characteristics ：

1. When the queue is full , Continuing to queue will block , Until another thread takes the element from the queue .

2. When the queue is empty , The queue will continue to block , Until another thread inserts an element into the queue .

---

2. One of the most important application scenarios in blocking queues  ： Producer consumer model

What is the producer consumer model ？？

【 give an example 】

  Follow the plan in the above figure to make two dumplings , Two things happen ：

1. The producer rolls the skin at an amazing speed , Not waiting for consumers to use , The curtain is full , So it stopped to have a rest .

2. The production speed of producers cannot keep up with the consumption speed of consumers , And then there was , There is no rolling in the curtain , So consumers wait for a rest .

We call this model producer consumer model ！！

Advantages of producer consumer model

Advantages one ： Can do better " Decoupling "

Advantage two ： Able to " Peak shaving and valley filling ", Improve the anti risk ability of the whole system ！！

  So we need to use producer consumer model to solve this problem ！！

---

 3.Java Simple use of blocking queues in the standard library

public class TestDemo1 {
    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();

        Thread customer = new Thread(() -> {
            while(true) {
                try {
                    int value = blockingQueue.take();
                    System.out.println(" Consumption elements : " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        customer.start();
        Thread producer = new Thread(() -> {
            int n = 0;
            while(true) {
                try {
                    System.out.println(" Production elements : " + n);
                    blockingQueue.put(n);
                    n++;
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        producer.start();
    }
}

---

4. Implementation of blocking queue

Based on array

Two core approaches ：put（ Queue entry ） and take（ Outgoing queue ）

【 Implementation code 】

public class MyBlockingQueue {
    // Assume that the maximum capacity of the array is  1000
    private int[] elem = new int[1000];
    // Team head 
    private int front = 0;
    // A party 
    private int rear = 0;
    // The number of valid data in the blocking queue 
    volatile private int size = 0;

    // Queue entry 
    public void put(int value) throws InterruptedException {
        synchronized (this) {
            while(size == elem.length) {
                // The queue is full , Queue up and wait 
                this.wait();
            }
            elem[rear] = value;
            rear++;
            if(rear == elem.length) {
                // If  rear  Reaching the end of the array , Let it start again  0  Start 
                rear = 0;
            }
            size++;
            // After inserting elements , You can put the   Outgoing queue   awakened 
            this.notify();
        }
    }
    // Outgoing queue 
    public Integer take() throws InterruptedException {
        int ret = 0;
        synchronized (this) {
            while(size == 0) {
                // The queue is empty , Queue up and wait 
                this.wait();
            }
            ret = this.elem[front];
            front++;
            if(front == elem.length) {
                front = 0;
            }
            size--;
            // After removing the element , You can put the   Queue entry   awakened 
            this.notify();
        }
        return ret;
    }
}

【 Basic steps 】

1. adopt " Circular queue " The way to achieve ;

2. Use synchronized Lock control ; 

3.put When inserting elements , Determine if the queue is full , it wait;

4.take When taking out elements , Determine if the queue is empty , it wait.

【 The key analysis 】

1. Out of the team , The team , And the number of elements ++,-- And so on are related to thread safety , Sentenced to empty 、 Full sentence , It's used again wait(), and wait() The first step of the method is to release the lock , therefore synchronized Wrap them all .

2. understand When the queue is empty , Waiting for leaving the team , After inserting elements , You can wake up and leave the team ; When the queue is full , Wait for the queue to be inserted , After taking out the elements , You can wake up and join the team

3. The entry and exit operations involve reading and writing , to size Variable plus volatile Prevent compiler optimization , Instruction reordering affects size Value .

4. Why? It is necessary to add a cycle when judging empty and full , Let's take a look at the picture below ：

【 Test code 】

public class Test {
    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue();
        Thread customer = new Thread(() -> {
            while(true) {
                try {
                    // Consumption elements 
                    int value = queue.take();
                    System.out.println(" consumption : " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        customer.start();
        Thread producer = new Thread(() -> {
            int value = 0;
            while(true) {
                try {
                    // Production elements 
                    queue.put(value);
                    System.out.println(" production : " + value);
                    value++;
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        producer.start();
    }
}

【 Share and master skills 】

1. Understand how to use ;

2. Understand the internal implementation principle ;

3. It can simulate and realize ;

4. Can tell others clearly .（ Writing a blog is telling others ）！！

---

That's all for this blog , Thank you for watching ！！