摘要：前言這篇主要來講解多線程中一個非常經典的設計模式包括它的基礎到拓展希望大家能夠有所收獲生產者消費者模式簡述此設計模式中主要分兩類線程生產者線程和消費者線程生產者提供數據和任務消費者處理數據和任務該模式的核心就是數據和任務的交互點共享內存緩

這篇主要來講解多線程中一個非常經典的設計模式
包括它的基礎到拓展
希望大家能夠有所收獲

此設計模式中主要分兩類線程：生產者線程和消費者線程
生產者提供數據和任務
消費者處理數據和任務
該模式的核心就是數據和任務的交互點：共享內存緩存區
下面給出簡單易懂的一張圖：

使用BlockingQueue來做緩沖區是非常合適的
通過BlockingQueue來理解生產者消費者模式
首先我們要知道BlockingQueue是什么？
它是一個實現接口，有很多實現類，比如:
ArrayBlockingQueue：前面講過，這個隊列適合做有界隊列，固定線程數
LinkedBlockingQueue：它適合做無界隊列
......

以ArrayBlockingQueue為例
它在內部放置了一個對象數組：

final Object[] items;

通過items數組來進行元素的存取
1(存).向隊列中壓入一個元素:
.offer()：如果隊列滿了，返回false
.put():將元素壓入隊列末尾，如果隊列滿了，它就會一直等待

2(取).向隊列中彈出元素（從頭部彈出）:
.poll()：如果隊列為空，返回null
.take()：如果隊列為空，繼續等待，知道隊列中有元素

了解了上面這些基礎后，我們來看下實際操作是怎樣的
在開始之前我們要有一個Entity類，只存一個long類型的value值進去：

public class MyData {
    private long value;

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}

有了這個數據模型，看下最后的執行main方法：

        public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();      //建立線程池
        BlockingQueue blockingQueue = new ArrayBlockingQueue(10);       //建立緩存隊列
        for (int i=0;i<3;i++){ Producer i = new Producer(queue); executor.execute(i); }      //制造三個生產線程
        for (int j=0;j<3;j++){ Consumer j = new Consumer(queue); executor.execute(j); }      //制造三個消費線程
        Thread.sleep(10000);
        for (int i=0;i<3;i++){ i.stop(); }      //停止生產
        Thread.sleep(5000);
        executor.shutdown();
        }

這里只給出Main，大家可以通過代碼簡單理解使用BlockingQueue做緩沖區的過程
沒有給出生產者和消費者的具體線程實現類，除了博主比較懶之外，還有是因為使用BlockingQueue做緩沖區并不推薦使用
雖然BlockingQueue是個不錯的選擇，但它使用了鎖和阻塞來保證線程間的同步，并不具備良好的并發性能
下面講解一種具有高性能的共享緩沖區

我們知道BlockingQueue隊列的性能不是特別優越
而之前講到過ConcurrentLinkedQueue是一個高性能隊列，因為它使用了大量的CAS操作
同理，如果我們利用CAS操作實現生產者-消費者模式，性能就可以得到客觀的提升
但是大量的CAS操作自己實現起來非常困難
所以推薦使用Disruptor框架

實際工作還是得使用成熟的框架，Disruptor是一款高效的無鎖內存隊列
它不像傳統隊列有head和tail指針來操控入列和出列
而是實現了一個固定大小的環形隊列(RingBuffer)，來看下實際模型圖：

生產者向緩沖區寫入數據，消費者從緩沖區讀取數據，大家都使用了CAS操作
而且由于是環形隊列的原因，可以做到完全的內存復用
從而大大減少系統分配空間以及回收空間的額外開銷

那么這個框架怎么使用呢？
1.導入包（博主使用了Maven依賴，不同版本大同小異）：

     com.lmax
     disruptor
     3.3.2

2.依舊創建一個entity類：

public class MyData {
    private long value;

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}

3.還要寫一個Factory類，細心的同學會看到環形隊列是固定大小的
這個Factory會在Disruptor實例對象構造時，構造所有緩沖區中的對象實例

public class DataFactory implements EventFactory{
    @Override
    public Object newInstance() {
        return new MyData();
    }
}

4.生產者（具體每行代碼的作用都已經注釋）：

public class Producers {
    private final RingBuffer ringBuffer;        //創建環形隊列(環形緩沖區)

    public Producers(RingBuffer ringBuffer) {
        this.ringBuffer = ringBuffer;           //將ringBuffer與Producers綁定
    }

    public void putData(ByteBuffer byteBuffer){         //此方法將產生的數據推入緩沖區

    long sequeue = ringBuffer.next();       //通過.next()方法得到ringBuffer的下一個節點，并且賦值給sequeue

    MyData event = ringBuffer.get(sequeue);     //將mydata數據存入到下一個節點

    event.setValue(byteBuffer.getLong(0));        //mydata的值有ByteBuffer參數帶入

    ringBuffer.publish(sequeue);        //將sequeue節點內的數據發布
    }
}

5.消費者：

public class Consumers implements WorkHandler{
    
    @Override
    public void onEvent(MyData myData) throws Exception {
        System.out.println("當前線程為:"+Thread.currentThread().getId()+"線程，它處理的數據是："+myData.getValue());
    }
}

6.執行函數：

public class RunTest {
        public static void main(String[] args) throws InterruptedException {
        Executor executor = Executors.newCachedThreadPool();        //創建線程池
        DataFactory dataFactory = new DataFactory();        //創建Factory實例
        int bufferSize = 1024;      //設置緩存區大小為1024（必須是2的整數次冪）
        Disruptor disruptor = new Disruptor(
                dataFactory,
                bufferSize,
                executor,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
                );
        disruptor.handleEventsWithWorkerPool(
                new Consumers(),
                new Consumers(),
                new Consumers(),
                new Consumers()
                );
        disruptor.start();      //Disruptor啟動
        RingBuffer ringBuffer = disruptor.getRingBuffer();      //實例化環形隊列并與Disruptor綁定
        Producers producers = new Producers(ringBuffer);        //實例化生產者并綁定ringBuffer
        ByteBuffer byteBuffe = ByteBuffer.allocate(8);      //創建一個容量為256字節的ByteBuffer
        for (long n = 0;true;n++){
            byteBuffe.putLong(0,n);
            producers.putData(byteBuffe);
            Thread.sleep(100);
            System.out.println("add data "+n);
        }
    }
}

我們來看下執行結果：

當前線程為:13線程，它處理的數據是：1059
add data 1059
當前線程為:11線程，它處理的數據是：1060
add data 1060
當前線程為:10線程，它處理的數據是：1061
add data 1061
當前線程為:12線程，它處理的數據是：1062
add data 1062
當前線程為:13線程，它處理的數據是：1063
add data 1063
當前線程為:11線程，它處理的數據是：1064
add data 1064
當前線程為:10線程，它處理的數據是：1065

可以看出，因為我無限的讓生產線程生產數據，而RingBuffer中那十幾條消費線程不停的消費數據

此外Disruptor不止CAS操作，還提供了四種等待策略讓消費者監控緩沖區的信息：
1.BlockingWaitStrategy：默認策略，最節省CPU，但在高并發下性能表現最糟糕
2.SleepingWaitStrategy：等待數據時自旋等待，不成功會使用LockSupport方法阻塞自己，通常用于異步日志
3.YieldWaitStrategy：用于低延時場合，在內部執行Thread.yield()死循環
4.BusySpinWaitStrategy：消費線程進行死循環監控緩沖區，吃掉所有CPU資源

除了CAS操作，消費者等待策略，Disruptor還使用CPU Cache的優化來進行優化

根據Disruptor官方報道：Disruptor的性能比BlockingQueuez至少高一倍以上！

以上便是生產者消費者模式的應用
謝謝閱讀，記得點關注看更新