多線程 - 生産者與消費者

生産者與消費者

線上程世界裡，生産者就是生産資料的線程，消費者就是消費資料的線程。在多線程開發當中，如果生産者處理速度很快，而消費者處理速度很慢，那麼生産者就必須等待消費者處理完，才能繼續生産資料。同樣的道理，如果消費者的處理能力大于生産者，那麼消費者就必須等待生産者。為了解決這種生産消費能力不均衡的問題，是以便有了生産者和消費者模式。

生産者消費者模式是通過一個容器來解決生産者和消費者的強耦合問題。生産者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，是以生産者生産完資料之後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生産者要資料，而是直接從阻塞隊列裡取，阻塞隊列就相當于一個緩沖區，平衡了生産者和消費者的處理能力。

這個阻塞隊列就是用來給生産者和消費者解耦的。縱觀大多數設計模式，都會找一個第三者出來進行解耦，如工廠模式的第三者是工廠類，模闆模式的第三者是模闆類。在學習一些設計模式的過程中，如果先找到這個模式的第三者，能幫助我們快速熟悉一個設計模式。

多線程 - 生産者與消費者

舉個例子：

在肯德基裡，你點單之後點單員會把所點的食物完成封裝之後拿來你面前，然後讓你結賬，有時候有些耗時操作沒完成就會留下一個餐台号稍後送來。

多線程 - 生産者與消費者

而在麥當勞的點餐模型大緻是，你點完快餐之後要求你立即付款，付完款之後下一位點餐，而取餐的是在旁邊等待，另一個服務員專責負責配餐。

多線程 - 生産者與消費者

在并發模型中，肯德基比較傾向于一個線程把所有的服務都做完，而麥當勞傾向于服務解耦，讓他們更專注于自己的業務。而肯德基的模型與BIO伺服器的模型設計類似，麥當勞的模型則與生産者消費者模型十分相似。

具體實作：

生産者負責生産一個數字并存入緩沖區，消費者從緩沖區中取出資料并且求出它的平方并輸出。

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生産者
 * @author ysm
 * 生産者消費者模型
 */

public class Producer implements Runnable {
    private volatile boolean isRunning = true;
    private BlockingQueue<PCData> queue;// 記憶體緩沖區
    private static AtomicInteger count = new AtomicInteger();// 總數 原子操作
    private static final int SLEEPTIME = 1000;

    public Producer(BlockingQueue<PCData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        PCData data = null;
        Random r = new Random();
        System.out.println("start producting id:" + Thread.currentThread().getId());
        try {
            while (isRunning) {
                Thread.sleep(r.nextInt(SLEEPTIME));
                data = new PCData(count.incrementAndGet());
                System.out.println(data + " 加入隊列");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.err.println(" 加入隊列失敗");
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }

    }

    public void stop() {
        isRunning = false;
    }
}

import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;

/**
 * 消費者
 * @author ysm
 */
public class Consumer implements Runnable{
    private BlockingQueue<PCData> queue;
    private static final int SLEEPTIME = 1000;
    public Consumer(BlockingQueue<PCData> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("start Consumer id :"+Thread.currentThread().getId());
        Random r = new Random();
        try{
            while(true){
                PCData data = queue.take();
                if(data != null)
                {
                    int re = data.getData() * data.getData();
                    System.out.println(MessageFormat.format("{0}*{1}={2}", data.getData(),data.getData(),re));
                    Thread.sleep(r.nextInt(SLEEPTIME));
                }
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 主函數
 * @author ysm
 *
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<PCData> queue = new LinkedBlockingDeque<>(10);
        Producer p1 = new Producer(queue);
        Producer p2 = new Producer(queue);
        Producer p3 = new Producer(queue);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
        Thread.sleep(10*1000);
        p1.stop();
        p2.stop();
        p3.stop();
        Thread.sleep(3000);
        service.shutdown();
    }
}

/**
 * 容器資料類型
 * @author ysm
 *
 */
public class PCData {
    private final int intData;
    public PCData(int d){
        intData = d;
    }
    public PCData(String d){
        intData = Integer.valueOf(d);
    }
    public int getData(){
        return intData;
    }
    @Override
    public String toString(){
        return "data:"+intData;
    }
}

因為BlockingQueue是一個阻塞隊列，它的存取可以保證隻有一個線程在進行，是以根據邏輯，生産者在記憶體滿的時候進行等待，并且喚醒消費者隊列，反過來消費者在饑餓狀态下等待并喚醒生産者進行生産。

下面的版本是使用notify/wait()方法進行設計的。在結構上是一緻遵從模型圖的。

import java.util.List;

/**
 * 消費者
 * 
 * @author ysm
 *
 */

public class Consumer implements Runnable {
    private List<PCData> queue;

    public Consumer(List<PCData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted())
                    break;
                PCData data = null;
                synchronized (queue) {
                    if (queue.size() == 0) {
                        queue.wait();
                        queue.notifyAll();
                    }
                    data = queue.remove(0);
                }
                System.out.println(
                        Thread.currentThread().getId() + " 消費了:" + data.get() + " result:" + (data.get() * data.get()));
                Thread.sleep(1000);
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import java.util.List;
import java.util.Random;

/**
 * 生産者
 * 
 * @author ysm
 *
 */
public class Producer implements Runnable {
    private List<PCData> queue;
    private int length;

    public Producer(List<PCData> queue, int length) {
        this.queue = queue;
        this.length = length;
    }

    @Override
    public void run() {
        try {
            while (true) {

                if (Thread.currentThread().isInterrupted())
                    break;
                Random r = new Random();
                long temp = r.nextInt(100);
                System.out.println(Thread.currentThread().getId() + " 生産了：" + temp);
                PCData data = new PCData();
                data.set(temp);
                synchronized (queue) {
                    if (queue.size() >= length) {
                        queue.notifyAll();
                        queue.wait();
                    } else
                        queue.add(data);
                }
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 測試主方法
 * @author ysm
 *
 */
public class Main {
    public static void main(String[] args) {
        List<PCData> queue = new ArrayList<PCData>();
        int length = 10;
        Producer p1 = new Producer(queue,length);
        Producer p2 = new Producer(queue,length);
        Producer p3 = new Producer(queue,length);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
        
    }
}

/**
 * 基本資料類型
 * @author ysm
 *
 */
public class PCData {
    private long value;
    public void set(long value){
        this.value = value;
        
    }
    public long get(){
        return value;
    }
}

多線程 - 生産者與消費者

下面的版本是使用await()/signal()方法進行設計的。在結構上是一緻遵從模型圖的。

import java.util.List;
/**
 * 消費者
 * @author ysm
 *
 */
public class Consumer implements Runnable{
    private List<PCData> queue;
    public Consumer(List<PCData> queue){
        this.queue = queue;
    }
    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted())
                    break;
                PCData data = null;
                Main.lock.lock();
                if (queue.size() == 0){
                    Main.full.signalAll();
                    Main.empty.await();
                }
                Thread.sleep(1000);
                data = queue.remove(0);
                Main.lock.unlock();
                System.out.println("消費者ID:"+Thread.currentThread().getId()+" 消費了:"+data.getData()+" result:"+(data.getData()*data.getData()));
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
}

import java.util.List;
import java.util.Random;
/**
 * 生産者
 * @author ysm
 *
 */
public class Producter implements Runnable{
    private List<PCData> queue;
    private int len;
    public Producter(List<PCData> queue,int len){
        this.queue = queue;
        this.len = len;
    }
    @Override
    public void run() {
        try{
            while(true){
                if(Thread.currentThread().isInterrupted())
                    break;
                Random r = new Random();
                PCData data = new PCData();
                data.setData(r.nextInt(500));
                Main.lock.lock();
                if(queue.size() >= len)
                {
                    Main.empty.signalAll();
                    Main.full.await();
                }
                Thread.sleep(1000);
                queue.add(data);
                Main.lock.unlock();
                System.out.println("生産者ID:"+Thread.currentThread().getId()+" 生産了:"+data.getData());
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
    public static ReentrantLock lock = new ReentrantLock();
    public static Condition empty = lock.newCondition();
    public static Condition full = lock.newCondition();
    public static void main(String[] args) {
        List<PCData> queue = new ArrayList<PCData>();
        int length = 10;
        Producter p1 = new Producter(queue,length);
        Producter p2 = new Producter(queue,length);
        Producter p3 = new Producter(queue,length);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);
        service.execute(p2);
        service.execute(p3);
        service.execute(c1);
        service.execute(c2);
        service.execute(c3);
    }
}

public class PCData {
    private int data;

    public int getData() {
        return data;
    }

    public void setData(int data) {
        this.data = data;
    }
}

多線程 - 生産者與消費者

Java中的線程池類其實就是一種生産者和消費者模式的實作方式，但是我覺得其實作方式更加高明。生産者把任務丢給線程池，線程池建立線程并處理任務，如果将要運作的任務數大于線程池的基本線程數就把任務扔到阻塞隊列裡，這種做法比隻使用一個阻塞隊列來實作生産者和消費者模式顯然要高明很多，因為消費者能夠處理直接就處理掉了，這樣速度更快，而生産者先存，消費者再取這種方式顯然慢一些。

我們的系統也可以使用線程池來實作多生産者消費者模式。比如建立N個不同規模的Java線程池來處理不同性質的任務，比如線程池1将資料讀到記憶體之後，交給線程池2裡的線程繼續處理壓縮資料。線程池1主要處理IO密集型任務，線程池2主要處理CPU密集型任務。

平時的工作中思考下哪些場景可以使用生産者消費者模式，這種場景應該非常之多，特别是需要處理任務時間比較長的場景，比如上傳附件并處理，使用者把檔案上傳到系統後，系統把檔案丢到隊列裡，然後立刻傳回告訴使用者上傳成功，最後消費者再去隊列裡取出檔案處理。比如調用一個遠端接口查詢資料，如果遠端服務接口查詢時需要幾十秒的時間，那麼它可以提供一個申請查詢的接口，這個接口把要申請查詢任務放資料庫中，然後該接口立刻傳回。然後伺服器端用線程輪詢并擷取申請任務進行處理，處理完之後發消息給調用方，讓調用方再來調用另外一個接口拿資料。

多線程 - 生産者與消費者

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method