并發編程中級篇三----并行設計模式----生產者-消費者模式

Aldous 發布于2019-08-16 10:47 / 1674人閱讀

摘要：生產者消費者模式是一個經典的多線程設計模式，它為多線程的協作提供了良好的解決方案。生產者消費者模式中的內存緩沖區的主要功能是數據在多線程間的共享。

生產者-消費者模式是一個經典的多線程設計模式，它為多線程的協作提供了良好的解決方案。在生產者-消費者模式中，通常有兩類線程，即若干個生產者線程和若干個消費者線程。生產者線程負責提交用戶請求，消費者線程負責處理用戶請求。生產者和消費者之間通過共享內存緩沖區進行通信。

生產者-消費者模式中的內存緩沖區的主要功能是數據在多線程間的共享。

1.創建一個被消費的對象

    public final class Data{
        
        private String id;
        
        private String name;
        
        //getter/setter(),toString()省略,構造方法省略
    }

2.創建一個生產者

    public class Provider implements Runnable{
        
        //共享緩存區
        private BlockingQueue queue;
        //多線程間是否啟動變量，有強制從主內存中刷新的功能。即時返回線程的狀態
        private volatile boolean isRunning = true;
        //id生成器
        private static AtomicInteger count = new AtomicInteger();
        //隨機對象
        private static Random r = new Random(); 
        
        public Provider(BlockingQueue queue){
            this.queue = queue;
        }
    
        @Override
        public void run() {
            while(isRunning){
                try {
                    //隨機休眠0 - 1000 毫秒 表示獲取數據(產生數據的耗時) 
                    Thread.sleep(r.nextInt(1000));
                    //獲取的數據進行累計...
                    int id = count.incrementAndGet();
                    //比如通過一個getData方法獲取了
                    Data data = new Data(Integer.toString(id), "數據" + id);
                    System.out.println("當前線程:" +
                     Thread.currentThread().getName() +
                      ", 獲取了數據，id為:" + id +
                       ", 進行裝載到公共緩沖區中...");
                    if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
                        System.out.println("提交緩沖區數據失敗....");
                        //do something... 比如重新提交
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        
        public void stop(){
            this.isRunning = false;
        }
    }

3.添加一個消費者

    public class Consumer implements Runnable{
        
        private BlockingQueue queue;
    
        public Consumer(BlockingQueue queue){
            this.queue = queue;
        }
        
        //隨機對象
        private static Random r = new Random(); 
    
        @Override
        public void run() {
            while(true){
                try {
                    //獲取數據
                    Data data = this.queue.take();
                    //進行數據處理。休眠0 - 1000毫秒模擬耗時
                    Thread.sleep(r.nextInt(1000));
                    System.out.println("當前消費線程：" + 
                    Thread.currentThread().getName() +
                     "， 消費成功，消費數據為id: " + data.getId());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

4.定義一個測試類

    public class Main{
        
        public static void main(String[] args) throws Exception {
            //內存緩沖區
            BlockingQueue queue = new LinkedBlockingQueue(10);
            //生產者
            Provider p1 = new Provider(queue);
            
            Provider p2 = new Provider(queue);
            Provider p3 = new Provider(queue);
            //消費者
            Consumer c1 = new Consumer(queue);
            Consumer c2 = new Consumer(queue);
            Consumer c3 = new Consumer(queue);
            //創建線程池運行,這是一個緩存的線程池，可以創建無窮大的線程，
            //沒有任務的時候不創建線程。空閑線程存活時間為60s（默認值）
    
            ExecutorService cachePool = Executors.newCachedThreadPool();
            cachePool.execute(p1);
            cachePool.execute(p2);
            cachePool.execute(p3);
            cachePool.execute(c1);
            cachePool.execute(c2);
            cachePool.execute(c3);
    
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            p1.stop();
            p2.stop();
            p3.stop();
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }        
    //        cachePool.shutdown(); 
    //        cachePool.shutdownNow();
            
    
        }
    }

運行結果如下所示

云服務器 GPU云服務器 java設計模式并發模式編程生產者消費者生產者消費者問題

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/70455.html

并發編程中級篇二----并行設計模式----Master-Wroker模式

摘要：模式是常用的并行計算模式，它的核心思想是系統是由兩類進程協助工作。負責接收和分配任務，負責處理子任務。當各個子進程處理完成后，會返回結果給，由做歸納和總結。其好處是能將一個大任務分解成若干個小任務，并行執行，從而提高系統的吞吐量。 Master-worker模式是常用的并行計算模式，它的核心思想是系統是由兩類進程協助工作。Master負責接收和分配任務，worker負責處理子任務。當各...

RebeccaZhong 2019-08-16 10:46 評論0 收藏0
并發編程中級篇二----并行設計模式----Future模式

摘要：模式類似于用戶提交商品訂單，下單成功以后后臺異步的執行耗時的業務在包中接口是線程模式的實現，可以來進行異步計算。 Future模式類似于用戶提交商品訂單，下單成功以后后臺異步的執行耗時的業務在java.util.concurrent包中.Future接口是Java線程Future模式的實現，可以來進行異步計算。 showImg(https://segmentfault.com/img/...

lx1036 2019-08-16 10:46 評論0 收藏0
你和阿里資深架構師之間，差的不僅僅是年齡（進階必看）

摘要：導讀閱讀本文需要有足夠的時間，筆者會由淺到深帶你一步一步了解一個資深架構師所要掌握的各類知識點，你也可以按照文章中所列的知識體系對比自身，對自己進行查漏補缺，覺得本文對你有幫助的話，可以點贊關注一下。目錄一基礎篇二進階篇三高級篇四架構篇五擴導讀：閱讀本文需要有足夠的時間，筆者會由淺到深帶你一步一步了解一個資深架構師所要掌握的各類知識點，你也可以按照文章中所列的知識體系對比自身，對自己...

huaixiaoz 2019-08-15 15:24 評論0 收藏0
Java 多線程編程基礎——Thread 類

摘要：程序執行時，至少會有一個線程在運行，這個運行的線程被稱為主線程。程序的終止是指除守護線程以外的線程全部終止。多線程程序由多個線程組成的程序稱為多線程程序。線程休眠期間可以被中斷，中斷將會拋出異常。線程我們在閱讀程序時，表面看來是在跟蹤程序的處理流程，實際上跟蹤的是線程的執行。單線程程序在單線程程序中，在某個時間點執行的處理只有一個。 Java 程序執行時，至少會有一個線程在運行...

zhoutk 2019-08-16 13:32 評論0 收藏0
并發（concurrency）與并行（parallellism）

摘要：并發與并行并發與并行的概念并行多個實例或者多臺機器同時執行一段處理邏輯，是真正的同時。并發通過調度算法，讓用戶看上去同時執行，實際上從操作層面不是真正的同時。并行與并發的異同點相似性都是為了合理且最大化利用系統的資源。并發（concurrency）與并行（parallellism）并發與并行的概念 ??并行：多個cpu實例或者多臺機器同時執行一段處理邏輯，是真正的同時。 ...

KavenFan 2019-08-19 11:05 評論0 收藏0