多執行緒設計模式Future、Master-Worker和生產者-消費者模型
阿新 • • 發佈：2019-01-01
並行設計模式屬於設計優化的一部分，它是對一些常用的多執行緒結構的總結和抽象。與序列結構相比，並行程式的結構通常更為複雜。因此合理的使用並行模式在多執行緒開發中更具有意義，在這裡主要介紹Future、Master-Worker和生產者-消費者模型。
Future模式
Future模式有點類似於商品訂單。比如在網購時，當看重某一件商品時，就可以提交訂單，當訂單處理完成後，在家裡等待商品送貨上門即可。或者說更形象的我們傳送ajax請求的時候，頁面是非同步的進行後臺處理，使用者無須一直等待請求的結果，可以繼續瀏覽或操作其他內容。
Future模式會非同步建立一個子執行緒，去完成相關請求任務，然後將處理結果返回給主執行緒main。在子執行緒請求並處理資料的過程中，主執行緒可以繼續做別的事情，即非同步載入資料。
Future模式非常適合在處理耗時很長的業務邏輯時進行使用，可以有效減少系統的響應時間，提高系統的吞吐量。
Main.java
FutureClient.javapublic class Main { public static void main(String[] args) throws InterruptedException { FutureClient fc = new FutureClient(); Data data = fc.request("請求引數"); System.out.println("請求傳送成功!"); System.out.println("做其他的事情..."); String result = data.getRequest(); System.out.println(result); } }
FutureData.javapublic class FutureClient { public Data request(final String queryStr){ //1 我想要一個代理物件（Data介面的實現類）先返回給傳送請求的客戶端，告訴他請求已經接收到，可以做其他的事情 final FutureData futureData = new FutureData(); //2 啟動一個新的執行緒，去載入真實的資料，傳遞給這個代理物件 new Thread(new Runnable() { @Override public void run() { //3 這個新的執行緒可以去慢慢的載入真實物件，然後傳遞給代理物件 RealData realData = new RealData(queryStr); futureData.setRealData(realData); } }).start(); return futureData; } }
public class FutureData implements Data{
private RealData realData ;
private boolean isReady = false;
public synchronized void setRealData(RealData realData) {
//如果已經裝載完畢了，就直接返回
if(isReady){
return;
}
//如果沒裝載，進行裝載真實物件
this.realData = realData;
isReady = true;
//進行通知
notify();
}
@Override
public synchronized String getRequest() {
//如果沒裝載好 程式就一直處於阻塞狀態
while(!isReady){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//裝載好直接獲取資料即可
return this.realData.getRequest();
}
}public class RealData implements Data{
private String result ;
public RealData (String queryStr){
System.out.println("根據" + queryStr + "進行查詢，這是一個很耗時的操作..");
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("操作完畢，獲取結果");
result = "查詢結果";
}
@Override
public String getRequest() {
return result;
}
}上述程式執行的過程：在main主執行緒中，建立FutureClient物件並呼叫其request方法。在request方法體執行過程中，返回一個FutureData代理物件給main主執行緒，同時開啟一個子執行緒A開始做真正請求處理工作。main主執行緒得到FutureData代理物件後，繼續向下執行程式碼，執行String result = data.getRequest();，即呼叫FutureData物件中的getRequest()方法，這個方法是一個同步方法，使用了synchronized關鍵字修飾，main主執行緒獲取當前FutureData的物件鎖之後，執行了wait()方法，main主執行緒處於阻塞狀態，並釋放了FutureData的物件鎖。在子執行緒A中建立RealData物件，執行RealData的建構函式，輸出內容並使當前執行緒休眠5S並繼續輸出內容。RealData物件建立完畢後，在子執行緒A中執行futureData.setRealData(realData);，這個方法同樣是一個同步方法，使用了synchronized關鍵字修飾，子執行緒A獲取到FutureData物件的鎖，執行notify(),發出通知，此時子執行緒A結束，阻塞的主執行緒main被喚醒，繼續執行getRequest()中的return this.realData.getRequest();。最後，主執行緒再次執行輸出。
Eclipse中console輸出如下：
JDK中Future模式的封裝
其實，在JDK中已經提供了Future模式的封裝，使用示例如下：
UseFuture.java
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
public class UseFuture implements Callable<String>{
private String para;
public UseFuture(String para){
this.para = para;
}
/**
* 這裡是真實的業務邏輯，其執行可能很慢
*/
@Override
public String call() throws Exception {
//模擬執行耗時
Thread.sleep(5000);
String result = this.para + "處理完成";
return result;
}
//主控制函式
public static void main(String[] args) throws Exception {
String queryStr = "query";
//構造FutureTask，並且傳入需要真正進行業務邏輯處理的類,該類一定是實現了Callable介面的類
FutureTask<String> future = new FutureTask<String>(new UseFuture(queryStr));
FutureTask<String> future2 = new FutureTask<String>(new UseFuture(queryStr));
//建立一個固定執行緒的執行緒池且執行緒數為2,
ExecutorService executor = Executors.newFixedThreadPool(2);
//這裡提交任務future,則開啟執行緒執行RealData的call()方法執行
//submit和execute的區別： 第一點是submit可以傳入實現Callable介面的例項物件， 第二點是submit方法有返回值
Future f1 = executor.submit(future);//單獨啟動一個執行緒A去執行
Future f2 = executor.submit(future2);//單獨啟動一個執行緒B去執行
System.out.println("請求完畢");
try {
//線上程A、B的執行過程中，主執行緒main可以做額外的資料操作，也就是主程式執行其他業務邏輯
System.out.println("處理實際的業務邏輯...");
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
//呼叫獲取資料方法,如果call()方法沒有執行完成,則依然會進行等待
System.out.println("資料：" + future.get());//future.get()獲取執行緒A執行任務的結果
System.out.println("資料：" + future2.get());//future2.get()獲取執行緒B執行任務的結果
executor.shutdown();
}
}Eclipse的console輸出：
Master-Worker模式
Master-Worker模式是常用的平行計算模式，它的核心思想是系統由兩類程序協作工作：Master程序和Worker程序。Master負責接收和分配任務，Worker負責處理子任務。當各個Worker子程序處理完成後，會將結果返回給Master，由Master做歸納和總結。其好處是能將一個大任務分解成若干個小任務，並行執行，從而提高系統的吞吐量。
在系統的資料量不是很大的場景，用Hadoop或者Storm有點大材小用，可以考慮Master-Worker。
Main.java
import java.util.Random;
public class Main {
public static void main(String[] args) {
System.out.println("本機器可用processor數量："+Runtime.getRuntime().availableProcessors());
Master master = new Master(new Worker(), Runtime.getRuntime().availableProcessors());
Random r = new Random();
for(int i = 1; i <= 100; i++){
Task t = new Task();
t.setId(i);
t.setPrice(r.nextInt(1000));
master.submit(t);
}
master.execute();
long start = System.currentTimeMillis();
while(true){
if(master.isComplete()){
long end = System.currentTimeMillis() - start;
int priceResult = master.getResult();
System.out.println("最終結果：" + priceResult + ", 執行時間：" + end);
break;
}
}
}
}import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
public class Master {
//1 有一個盛放任務的容器
private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
//2 需要有一個盛放worker的集合
private HashMap<String, Thread> workers = new HashMap<String, Thread>();
//3 需要有一個盛放每一個worker執行任務的結果集合
private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
//4 構造方法
public Master(Worker worker , int workerCount){
worker.setWorkQueue(this.workQueue);
worker.setResultMap(this.resultMap);
for(int i = 0; i < workerCount; i ++){
this.workers.put(Integer.toString(i), new Thread(worker));
}
}
//5 需要一個提交任務的方法
public void submit(Task task){
this.workQueue.add(task);
}
//6 需要有一個執行的方法，啟動所有的worker方法去執行任務
public void execute(){
for(Map.Entry<String, Thread> me : workers.entrySet()){
me.getValue().start();
}
}
//7 判斷是否執行結束的方法
public boolean isComplete() {
for(Map.Entry<String, Thread> me : workers.entrySet()){
if(me.getValue().getState() != Thread.State.TERMINATED){
return false;
}
}
return true;
}
//8 計算結果方法
public int getResult() {
int priceResult = 0;
for(Map.Entry<String, Object> me : resultMap.entrySet()){
priceResult += (Integer)me.getValue();
}
return priceResult;
}
}public class Task {
private int id;
private int price ;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public int getPrice() {
return price;
}
public void setPrice(int price) {
this.price = price;
}
}import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
public class Worker implements Runnable {
private ConcurrentLinkedQueue<Task> workQueue;
private ConcurrentHashMap<String, Object> resultMap;
public void setWorkQueue(ConcurrentLinkedQueue<Task> workQueue) {
this.workQueue = workQueue;
}
public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
this.resultMap = resultMap;
}
@Override
public void run() {
while(true){
Task input = this.workQueue.poll();//獲取並移除佇列的頭元素
if(input == null) break;
Object output = MyWorker.handle(input);
this.resultMap.put(Integer.toString(input.getId()), output);
}
}
/*private Object handle(Task input) {
Object output = null;
try {
//處理任務的耗時。。 比如說進行操作資料庫。。。
Thread.sleep(500);
output = input.getPrice();
} catch (InterruptedException e) {
e.printStackTrace();
}
return output;
}*/
private static Object handle(Task input) {
return null;
}
}import test.Worker;
public class MyWorker extends Worker{
public static Object handle(Task input){
Object output = null;
try {
//處理任務的耗時。。 比如說進行操作資料庫。。。
Thread.sleep(500);
output = input.getPrice();
} catch (InterruptedException e) {
e.printStackTrace();
}
return output;
}
}
生產者-消費者
生產者-消費者也是一個非常經典的多執行緒模式，我們在實際開發中應用非常廣泛的的思想理念。在生產者-消費者模式中：通常有兩類執行緒，即若干個生產者的執行緒和若干個消費者的執行緒。生產者執行緒負責提交使用者請求，消費者執行緒負責具體處理生產者提交的任務，在生產者和消費者之間通過共享記憶體快取區進行通訊。
Main.java
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
public static void main(String[] args) throws Exception {
//記憶體緩衝區，生產者和消費者都需要擁有記憶體緩衝區的引用
BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(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();
//將3個生產者、3個消費者交給執行緒池去執行，執行緒池會分配執行緒去執行這些生產者、消費者的任務
cachePool.execute(p1);//execute方法的引數為實現了Runnable介面的類的物件
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();
}
}import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Provider implements Runnable{
//共享快取區
private BlockingQueue<Data> 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;
}
}public final class Data {
private String id;
private String name;
public Data(String id, String name){
this.id = id;
this.name = name;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString(){
return "{id: " + id + ", name: " + name + "}";
}
}import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class Consumer implements Runnable{
private BlockingQueue<Data> 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();
}
}
}
}Eclipse的console中輸出：
