摘要：生產(chǎn)者消費者模式是一個經(jīng)典的多線程設計模式，它為多線程的協(xié)作提供了良好的解決方案。生產(chǎn)者消費者模式中的內(nèi)存緩沖區(qū)的主要功能是數(shù)據(jù)在多線程間的共享。

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

生產(chǎn)者-消費者模式中的內(nèi)存緩沖區(qū)的主要功能是數(shù)據(jù)在多線程間的共享。

1.創(chuàng)建一個被消費的對象

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

2.創(chuàng)建一個生產(chǎn)者

    public class Provider implements Runnable{
        
        //共享緩存區(qū)
        private BlockingQueue queue;
        //多線程間是否啟動變量，有強制從主內(nèi)存中刷新的功能。即時返回線程的狀態(tài)
        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 毫秒 表示獲取數(shù)據(jù)(產(chǎn)生數(shù)據(jù)的耗時) 
                    Thread.sleep(r.nextInt(1000));
                    //獲取的數(shù)據(jù)進行累計...
                    int id = count.incrementAndGet();
                    //比如通過一個getData方法獲取了
                    Data data = new Data(Integer.toString(id), "數(shù)據(jù)" + id);
                    System.out.println("當前線程:" +
                     Thread.currentThread().getName() +
                      ", 獲取了數(shù)據(jù)，id為:" + id +
                       ", 進行裝載到公共緩沖區(qū)中...");
                    if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
                        System.out.println("提交緩沖區(qū)數(shù)據(jù)失敗....");
                        //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 {
                    //獲取數(shù)據(jù)
                    Data data = this.queue.take();
                    //進行數(shù)據(jù)處理。休眠0 - 1000毫秒模擬耗時
                    Thread.sleep(r.nextInt(1000));
                    System.out.println("當前消費線程：" + 
                    Thread.currentThread().getName() +
                     "， 消費成功，消費數(shù)據(jù)為id: " + data.getId());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

4.定義一個測試類

    public class Main{
        
        public static void main(String[] args) throws Exception {
            //內(nèi)存緩沖區(qū)
            BlockingQueue queue = new LinkedBlockingQueue(10);
            //生產(chǎn)者
            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);
            //創(chuàng)建線程池運行,這是一個緩存的線程池，可以創(chuàng)建無窮大的線程，
            //沒有任務的時候不創(chuàng)建線程。空閑線程存活時間為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();
            
    
        }
    }

運行結(jié)果如下所示