快捷導(dǎo)航

Java使用阻塞隊列BlockingQueue實現(xiàn)生產(chǎn)者消費者的方法

更新時間：2024年11月11日 14:11:06 作者：麥田

BlockingQueue是一個支持阻塞插入和移除操作的隊列,常用于多線程環(huán)境下的生產(chǎn)者和消費者場景,文章介紹了阻塞隊列BlockingQueue的概念和其在生產(chǎn)者消費者模式中的應(yīng)用,提供了一個簡單的示例,展示了如何使用ArrayBlockingQueue來實現(xiàn)生產(chǎn)者消費者模式

什么是阻塞隊列

阻塞隊列(BlockingQueue)是一個支持兩個附加操作的隊列。這兩個附加的操作支持阻塞的插入和移除方法。

1、支持阻塞的插入方法：意思是當(dāng)隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。
2、支持阻塞的移除方法：意思是在隊列為空時，獲取元素的線程會等待隊列變?yōu)榉强铡?/li>

阻塞隊列常用于生產(chǎn)者和消費者的場景，生產(chǎn)者是向隊列里添加元素的線程，消費者是從隊列里取元素的線程。阻塞隊列就是生產(chǎn)者用來存放元素、消費者用來獲取元素的容器。

Java中提供了幾個對BlockingQueue的實現(xiàn)類，如: ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue 等

在處理生產(chǎn)者/消費者問題上我們將會使用ArrayBlockingQueue來實現(xiàn)，如下是我們需知道的重要方法：

put(E e): 這個方法用于向隊列中插入元素，如果隊列已滿，需要等待可用的這間。
E take(): 這個方法用于從隊列頭部獲取或者移除元素，如果隊列為空則需要等待可用的元素。

使用BlockingQueue來解決生產(chǎn)者/消費者示例

Mantou類

Producer產(chǎn)生的普通Java對象，并添加到隊列中。

/**
 * Producer產(chǎn)生的饅頭類
 * @author itmyhome
 *
 */
public class Mantou {
    private String mantou;
    public Mantou(String mantou) {
        this.mantou = mantou;
    }
    public String getMantou() {
        return mantou;
    }
    public void setMantou(String mantou) {
        this.mantou = mantou;
    }
}

Producer生產(chǎn)者類

Producer這個類會產(chǎn)生消息并將其放入隊列中。

import java.util.concurrent.BlockingQueue;
public class Producer implements Runnable {
    BlockingQueue<Mantou> queue;
    public Producer(BlockingQueue<Mantou> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        // 生產(chǎn)饅頭
        for (int i = 0; i < 100; i++) {
            Mantou mt = new Mantou("" + i);
            try {
                Thread.sleep(100);
                queue.put(mt);
                System.out.println("生產(chǎn)饅頭: " + mt.getMantou());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 添加退出消息
        Mantou msg = new Mantou("exit");
        try {
            queue.put(msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Consumer消費者類

Consumer類會從隊列獲取消息進行處理。如果獲取的是退出消息則結(jié)束。

import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
    BlockingQueue<Mantou> queue;
    public Consumer(BlockingQueue<Mantou> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            Mantou mantou;
            // 獲取并處理消息直到接收到“exit”消息
            while (!(mantou = queue.take()).getMantou().equals("exit")) {
                Thread.sleep(100);
                System.out.println("消費饅頭： " + mantou.getMantou());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

ProducerConsumerService

生產(chǎn)者/消費者的服務(wù)類將會產(chǎn)生固定大小的BlockingQueue，生產(chǎn)者和消費者同時共享該BlockingQueue，該服務(wù)類會起啟動生產(chǎn)者和消費者線程。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * @author itmyhome
 *
 */
public class ProducerConsumerService {
    public static void main(String[] args) {
        // 創(chuàng)建大小為10的 BlockingQueue
        BlockingQueue<Mantou> queue = new ArrayBlockingQueue<Mantou>(10);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);
        // 開啟 producer線程向隊列中生產(chǎn)消息
        new Thread(producer).start();
        //開啟 consumer線程 中隊列中消費消息
        new Thread(consumer).start();
        System.out.println("Producer and Consumer has been started");
    }
}

程序運行結(jié)果：

Producer and Consumer has been started
生產(chǎn)饅頭： 0
生產(chǎn)饅頭： 1
消費饅頭： 0
消費饅頭： 1
生產(chǎn)饅頭： 2
消費饅頭： 2
生產(chǎn)饅頭： 3
消費饅頭： 3
生產(chǎn)饅頭： 4
消費饅頭： 4
生產(chǎn)饅頭： 5
消費饅頭： 5
生產(chǎn)饅頭： 6
消費饅頭： 6
......