在本Java教程中，動力節(jié)點小編將通過相同的生產(chǎn)者/消費者概念來解釋BlockingQueue in Java.

Java中阻塞隊列的優(yōu)點是什么?

Ajava.util.Queue 支持在檢索元素時等待隊列變?yōu)榉强?，并在存儲元素時等待隊列中的空間變?yōu)榭捎玫牟僮鳌?/p>

我們需要創(chuàng)建四個 Java 類：

CrunchifyMessage.java 放置和獲取消息

CrunchifyBlockingProducer.java將消息放入隊列

CrunchifyBlockingConsumer.java 從隊列中獲取消息

CrunchifyBlockingMain.java 開始測試

BlockingQueue 實現(xiàn)是thread-safe. 所有排隊方法本質(zhì)上都是原子的并且使用內(nèi)部鎖。

讓我們開始在 Java 中實現(xiàn)線程安全的 BlockingQueue

第1步

創(chuàng)建類 CrunchifyMessage.java。這是簡單的Java 對象。

package com.crunchify.example;
/**
 * @author Crunchify.com 
 * simple Message class to put and get message into queue
 */
public class CrunchifyMessage {
    private String crunchifyMsg;    
    public CrunchifyMessage(String string) {
        this.crunchifyMsg = string;
    }    
    public String getMsg() {
        return crunchifyMsg;
    }
}

第2步

CrunchifyBlockingProducer.java 創(chuàng)建創(chuàng)建簡單味精并將其放入隊列的生產(chǎn)者 。

package com.crunchify.example;
import java.util.concurrent.BlockingQueue;
/**
 * @author Crunchify.com
 * 
 */
public class CrunchifyBlockingProducer implements Runnable {  
    private BlockingQueue<CrunchifyMessage> crunchQueue;    
    public CrunchifyBlockingProducer(BlockingQueue<CrunchifyMessage> queue) {
        this.crunchQueue = queue;
    }    
    @Override
    public void run() {
        // producing CrunchifyMessage messages
        for (int i = 1; i <= 5; i++) {
            CrunchifyMessage msg = new CrunchifyMessage("i'm msg " + i);
            try {
                Thread.sleep(10);
                crunchQueue.put(msg);
                System.out.println("CrunchifyBlockingProducer: Message - " + msg.getMsg() + " produced.");
            } catch (Exception e) {
                System.out.println("Exception:" + e);
            }
        }        
        // adding exit message
        CrunchifyMessage msg = new CrunchifyMessage("All done from Producer side. Produced 50 CrunchifyMessages");
        try {
            crunchQueue.put(msg);
            System.out.println("CrunchifyBlockingProducer: Exit Message - " + msg.getMsg());
        } catch (Exception e) {
            System.out.println("Exception:" + e);
        }
    }   
}

第3步

創(chuàng)建 CrunchifyBlockingConsumer.java 從隊列中消費消息的類。

package com.crunchify.example;
import java.util.concurrent.BlockingQueue;
/**
 * @author Crunchify.com
 * 
 */ 
public class CrunchifyBlockingConsumer implements Runnable {   
    private BlockingQueue<CrunchifyMessage> queue;    
    public CrunchifyBlockingConsumer(BlockingQueue<CrunchifyMessage> queue) {
        this.queue = queue;
    }    
    @Override
    public void run() {
        try {
            CrunchifyMessage msg;            
            // consuming messages until exit message is received
            while ((msg = queue.take()).getMsg() != "exit") {
                Thread.sleep(10);
                System.out.println("CrunchifyBlockingConsumer: Message - " + msg.getMsg() + " consumed.");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }   
}

第4步

創(chuàng)建 CrunchifyBlockingMain.java 運行 BlockingQueue測試的簡單方法。運行這個程序來檢查 BlockingQueue 的行為。

package com.crunchify.example;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue; 
/**
 * @author Crunchify.com
 * 
 */
public class CrunchifyBlockingMain {   
    public static void main(String[] args) {       
        // Creating BlockingQueue of size 10
        // BlockingQueue supports operations that wait for the queue to become non-empty when retrieving an element, and
        // wait for space to become available in the queue when storing an element.
        BlockingQueue<CrunchifyMessage> crunchQueue = new ArrayBlockingQueue<>(10);
        CrunchifyBlockingProducer crunchProducer = new CrunchifyBlockingProducer(crunchQueue);
        CrunchifyBlockingConsumer crunchConsumer = new CrunchifyBlockingConsumer(crunchQueue);        
        // starting producer to produce messages in queue
        new Thread(crunchProducer).start();        
        // starting consumer to consume messages from queue
        new Thread(crunchConsumer).start();        
        System.out.println("Let's get started. Producer / Consumer Test Started.\n");
    }   
}

BlockingQueue不接受空元素。在嘗試添加、放置或提供null時，實現(xiàn)會拋出NullPointerException。

null用作標記值以指示輪詢操作失敗。

結(jié)果：

Let's get started. Producer / Consumer Test Started.
CrunchifyBlockingProducer: Message - i'm msg 1 produced.
CrunchifyBlockingProducer: Message - i'm msg 2 produced.
CrunchifyBlockingConsumer: Message - i'm msg 1 consumed.
CrunchifyBlockingConsumer: Message - i'm msg 2 consumed.
CrunchifyBlockingProducer: Message - i'm msg 3 produced.
CrunchifyBlockingConsumer: Message - i'm msg 3 consumed.
CrunchifyBlockingProducer: Message - i'm msg 4 produced.
CrunchifyBlockingConsumer: Message - i'm msg 4 consumed.
CrunchifyBlockingProducer: Message - i'm msg 5 produced.
CrunchifyBlockingProducer: Exit Message - All done from Producer side. Produced 50 CrunchifyMessages
CrunchifyBlockingConsumer: Message - i'm msg 5 consumed.
CrunchifyBlockingConsumer: Message - All done from Producer side. Produced 50 CrunchifyMessages consumed.

什么時候應(yīng)該使用 java.util.concurrent.BlockingQueue?

當您想限制某種傳入請求時，您應(yīng)該使用相同的

生產(chǎn)者可以通過無限隊列遠遠領(lǐng)先于消費者。如果消費者沒有趕上生產(chǎn)者，那么它可能會導致 OutOfMemoryError. 在這種情況下，最好向潛在的生產(chǎn)者發(fā)出隊列已滿的信號，并在失敗后迅速放棄。

換句話說：生產(chǎn)者自然受到限制。

阻塞隊列通常用于并發(fā)應(yīng)用程序

它提供了正確的、線程安全的實現(xiàn)

內(nèi)存消耗也應(yīng)該受到限制