點(diǎn)擊上方“碼農(nóng)突圍”，馬上關(guān)注
這里是碼農(nóng)充電第一站，回復(fù)“666”，獲取一份專屬大禮包
真愛，請(qǐng)?jiān)O(shè)置“星標(biāo)”或點(diǎn)個(gè)“在看

作者：大道方圓

cnblogs.com/xdecode/p/9137793.html

本文主要講解幾種常見并行模式, 具體目錄結(jié)構(gòu)如下圖.

單例

單例是最常見的一種設(shè)計(jì)模式, 一般用于全局對(duì)象管理, 比如xml配置讀寫之類的.

一般分為懶漢式, 餓漢式.

懶漢式: 方法上加synchronized

public?static?synchronized?Singleton?getInstance()?{??
?????????if?(single?==?null)?{????
?????????????single?=?new?Singleton();??
?????????}????
????????return?single;??
}??

這種方式, 由于每次獲取示例都要獲取鎖, 不推薦使用, 性能較差

懶漢式: 使用雙檢鎖 + volatile

private?volatile?Singleton?singleton?=?null;
????public?static?Singleton?getInstance()?{
????????if?(singleton?==?null)?{
????????????synchronized?(Singleton.class)?{
????????????????if?(singleton?==?null)?{
????????????????????singleton?=?new?Singleton();
????????????????}
????????????}
????????}
????????return?singleton;
????}

本方式是對(duì)直接在方法上加鎖的一個(gè)優(yōu)化, 好處在于只有第一次初始化獲取了鎖.

后續(xù)調(diào)用getInstance已經(jīng)是無鎖狀態(tài). 只是寫法上稍微繁瑣點(diǎn).

至于為什么要volatile關(guān)鍵字, 主要涉及到j(luò)dk指令重排, 詳見之前的博文: Java內(nèi)存模型與指令重排

懶漢式: 使用靜態(tài)內(nèi)部類

public?class?Singleton?{
????private?static?class?LazyHolder?{
???????private?static?final?Singleton?INSTANCE?=?new?Singleton();
????}
????private?Singleton?(){}
????public?static?final?Singleton?getInstance()?{
???????return?LazyHolder.INSTANCE;
????}
}

該方式既解決了同步問題, 也解決了寫法繁瑣問題. 推薦使用改寫法.

缺點(diǎn)在于無法響應(yīng)事件來重新初始化INSTANCE.

餓漢式

public?class?Singleton1?{
????private?Singleton1()?{}
????private?static?final?Singleton1?single?=?new?Singleton1();
????public?static?Singleton1?getInstance()?{
????????return?single;
????}
}

缺點(diǎn)在于對(duì)象在一開始就直接初始化了.

Future模式

該模式的核心思想是異步調(diào)用. 有點(diǎn)類似于異步的ajax請(qǐng)求.

當(dāng)調(diào)用某個(gè)方法時(shí), 可能該方法耗時(shí)較久, 而在主函數(shù)中也不急于立刻獲取結(jié)果.

因此可以讓調(diào)用者立刻返回一個(gè)憑證, 該方法放到另外線程執(zhí)行,后續(xù)主函數(shù)拿憑證再去獲取方法的執(zhí)行結(jié)果即可, 其結(jié)構(gòu)圖如下

jdk中內(nèi)置了Future模式的支持, 其接口如下:

通過FutureTask實(shí)現(xiàn)

注意其中兩個(gè)耗時(shí)操作.

• 如果doOtherThing耗時(shí)2s, 則整個(gè)函數(shù)耗時(shí)2s左右.

• 如果doOtherThing耗時(shí)0.2s, 則整個(gè)函數(shù)耗時(shí)取決于RealData.costTime, 即1s左右結(jié)束.

public?class?FutureDemo1?{

????public?static?void?main(String[]?args)?throws?InterruptedException,?ExecutionException?{
????????FutureTask?future?=?new?FutureTask(new?Callable()?{
????????????@Override
????????????public?String?call()?throws?Exception?{
????????????????return?new?RealData().costTime();
????????????}
????????});
????????ExecutorService?service?=?Executors.newCachedThreadPool();
????????service.submit(future);

????????System.out.println("RealData方法調(diào)用完畢");
????????//?模擬主函數(shù)中其他耗時(shí)操作
????????doOtherThing();
????????//?獲取RealData方法的結(jié)果
????????System.out.println(future.get());
????}

????private?static?void?doOtherThing()?throws?InterruptedException?{
????????Thread.sleep(2000L);
????}
}

class?RealData?{

????public?String?costTime()?{
????????try?{
????????????//?模擬RealData耗時(shí)操作
????????????Thread.sleep(1000L);
????????????return?"result";
????????}?catch?(InterruptedException?e)?{
????????????e.printStackTrace();
????????}
????????return?"exception";
????}

}

通過Future實(shí)現(xiàn)

與上述FutureTask不同的是, RealData需要實(shí)現(xiàn)Callable接口.

public?class?FutureDemo2?{

????public?static?void?main(String[]?args)?throws?InterruptedException,?ExecutionException?{
????????ExecutorService?service?=?Executors.newCachedThreadPool();
????????Future?future?=?service.submit(new?RealData2());

????????System.out.println("RealData2方法調(diào)用完畢");
????????//?模擬主函數(shù)中其他耗時(shí)操作
????????doOtherThing();
????????//?獲取RealData2方法的結(jié)果
????????System.out.println(future.get());
????}

????private?static?void?doOtherThing()?throws?InterruptedException?{
????????Thread.sleep(2000L);
????}
}

class?RealData2?implements?Callable<String>{

????public?String?costTime()?{
????????try?{
????????????//?模擬RealData耗時(shí)操作
????????????Thread.sleep(1000L);
????????????return?"result";
????????}?catch?(InterruptedException?e)?{
????????????e.printStackTrace();
????????}
????????return?"exception";
????}

????@Override
????public?String?call()?throws?Exception?{
????????return?costTime();
????}
}

另外Future本身還提供了一些額外的簡單控制功能, 其API如下

//?取消任務(wù)
????boolean?cancel(boolean?mayInterruptIfRunning);
????//?是否已經(jīng)取消
????boolean?isCancelled();
????//?是否已經(jīng)完成
????boolean?isDone();
????//?取得返回對(duì)象
????V?get()?throws?InterruptedException,?ExecutionException;
????//?取得返回對(duì)象,?并可以設(shè)置超時(shí)時(shí)間
????V?get(long?timeout,?TimeUnit?unit)
????????????throws?InterruptedException,?ExecutionException,?TimeoutException;

生產(chǎn)消費(fèi)者模式

生產(chǎn)者-消費(fèi)者模式是一個(gè)經(jīng)典的多線程設(shè)計(jì)模式. 它為多線程間的協(xié)作提供了良好的解決方案。

在生產(chǎn)者-消費(fèi)者模式中，通常由兩類線程，即若干個(gè)生產(chǎn)者線程和若干個(gè)消費(fèi)者線程。

生產(chǎn)者線程負(fù)責(zé)提交用戶請(qǐng)求，消費(fèi)者線程則負(fù)責(zé)具體處理生產(chǎn)者提交的任務(wù)。

生產(chǎn)者和消費(fèi)者之間則通過共享內(nèi)存緩沖區(qū)進(jìn)行通信, 其結(jié)構(gòu)圖如下

PCData為我們需要處理的元數(shù)據(jù)模型, 生產(chǎn)者構(gòu)建PCData, 并放入緩沖隊(duì)列.

消費(fèi)者從緩沖隊(duì)列中獲取數(shù)據(jù), 并執(zhí)行計(jì)算.

生產(chǎn)者核心代碼

while(isRunning)?{
????????????Thread.sleep(r.nextInt(SLEEP_TIME));
????????????data?=?new?PCData(count.incrementAndGet);
????????????//?構(gòu)造任務(wù)數(shù)據(jù)
????????????System.out.println(data?+?"?is?put?into?queue");
????????????if?(!queue.offer(data,?2,?TimeUnit.SECONDS))?{
????????????????//?將數(shù)據(jù)放入隊(duì)列緩沖區(qū)中
????????????????System.out.println("faild?to?put?data?:?"?+?data);
????????????}
????????}

消費(fèi)者核心代碼

while?(true)?{
????????????PCData?data?=?queue.take();
????????????//?提取任務(wù)
????????????if?(data?!=?null)?{
????????????????//?獲取數(shù)據(jù),?執(zhí)行計(jì)算操作
????????????????int?re?=?data.getData()?*?10;
????????????????System.out.println("after?cal,?value?is?:?"?+?re);
????????????????Thread.sleep(r.nextInt(SLEEP_TIME));
????????????}
????????}

生產(chǎn)消費(fèi)者模式可以有效對(duì)數(shù)據(jù)解耦, 優(yōu)化系統(tǒng)結(jié)構(gòu).

降低生產(chǎn)者和消費(fèi)者線程相互之間的依賴與性能要求.

一般使用BlockingQueue作為數(shù)據(jù)緩沖隊(duì)列, 他是通過鎖和阻塞來實(shí)現(xiàn)數(shù)據(jù)之間的同步,

如果對(duì)緩沖隊(duì)列有性能要求, 則可以使用基于CAS無鎖設(shè)計(jì)的ConcurrentLinkedQueue.

分而治之

嚴(yán)格來講, 分而治之不算一種模式, 而是一種思想.

它可以將一個(gè)大任務(wù)拆解為若干個(gè)小任務(wù)并行執(zhí)行, 提高系統(tǒng)吞吐量.

我們主要講兩個(gè)場景, Master-Worker模式, ForkJoin線程池.

Master-Worker模式

該模式核心思想是系統(tǒng)由兩類進(jìn)行協(xié)助工作: Master進(jìn)程, Worker進(jìn)程.

Master負(fù)責(zé)接收與分配任務(wù), Worker負(fù)責(zé)處理任務(wù). 當(dāng)各個(gè)Worker處理完成后,

將結(jié)果返回給Master進(jìn)行歸納與總結(jié).

假設(shè)一個(gè)場景, 需要計(jì)算100個(gè)任務(wù), 并對(duì)結(jié)果求和, Master持有10個(gè)子進(jìn)程.

Master代碼

public?class?MasterDemo?{
????//?盛裝任務(wù)的集合
????private?ConcurrentLinkedQueue?workQueue?=?new?ConcurrentLinkedQueue();
????//?所有worker
????private?HashMap?workers?=?new?HashMap<>();
????//?每一個(gè)worker并行執(zhí)行任務(wù)的結(jié)果
????private?ConcurrentHashMap?resultMap?=?new?ConcurrentHashMap<>();

????public?MasterDemo(WorkerDemo?worker,?int?workerCount)?{
????????//?每個(gè)worker對(duì)象都需要持有queue的引用,?用于領(lǐng)任務(wù)與提交結(jié)果
????????worker.setResultMap(resultMap);
????????worker.setWorkQueue(workQueue);
????????for?(int?i?=?0;?i?????????????workers.put("子節(jié)點(diǎn):?"?+?i,?new?Thread(worker));
????????}
????}

????//?提交任務(wù)
????public?void?submit(TaskDemo?task)?{
????????workQueue.add(task);
????}

????//?啟動(dòng)所有的子任務(wù)
????public?void?execute(){
????????for?(Map.Entry?entry?:?workers.entrySet())?{
????????????entry.getValue().start();
????????}
????}

????//?判斷所有的任務(wù)是否執(zhí)行結(jié)束
????public?boolean?isComplete()?{
????????for?(Map.Entry?entry?:?workers.entrySet())?{
????????????if?(entry.getValue().getState()?!=?Thread.State.TERMINATED)?{
????????????????return?false;
????????????}
????????}

????????return?true;
????}

????//?獲取最終匯總的結(jié)果
????public?int?getResult()?{
????????int?result?=?0;
????????for?(Map.Entry?entry?:?resultMap.entrySet())?{
????????????result?+=?Integer.parseInt(entry.getValue().toString());
????????}

????????return?result;
????}

}

Worker代碼

public?class?WorkerDemo?implements?Runnable{

????private?ConcurrentLinkedQueue?workQueue;
????private?ConcurrentHashMap?resultMap;

????@Override
????public?void?run()?{
????????while?(true)?{
????????????TaskDemo?input?=?this.workQueue.poll();
????????????//?所有任務(wù)已經(jīng)執(zhí)行完畢
????????????if?(input?==?null)?{
????????????????break;
????????????}
????????????//?模擬對(duì)task進(jìn)行處理,?返回結(jié)果
????????????int?result?=?input.getPrice();
????????????this.resultMap.put(input.getId()?+?"",?result);
????????????System.out.println("任務(wù)執(zhí)行完畢,?當(dāng)前線程:?"?+?Thread.currentThread().getName());
????????}
????}

????public?ConcurrentLinkedQueue?getWorkQueue()?{
????????return?workQueue;
????}

????public?void?setWorkQueue(ConcurrentLinkedQueue?workQueue)?{
????????this.workQueue?=?workQueue;
????}

????public?ConcurrentHashMap?getResultMap()?{
????????return?resultMap;
????}

????public?void?setResultMap(ConcurrentHashMap?resultMap)?{
????????this.resultMap?=?resultMap;
????}
}

public?class?TaskDemo?{

????private?int?id;
????private?String?name;
????private?int?price;

????public?int?getId()?{
????????return?id;
????}

????public?void?setId(int?id)?{
????????this.id?=?id;
????}

????public?String?getName()?{
????????return?name;
????}

????public?void?setName(String?name)?{
????????this.name?=?name;
????}

????public?int?getPrice()?{
????????return?price;
????}

????public?void?setPrice(int?price)?{
????????this.price?=?price;
????}
}

主函數(shù)測試

MasterDemo?master?=?new?MasterDemo(new?WorkerDemo(),?10);
????????for?(int?i?=?0;?i?100;?i++)?{
????????????TaskDemo?task?=?new?TaskDemo();
????????????task.setId(i);
????????????task.setName("任務(wù)"?+?i);
????????????task.setPrice(new?Random().nextInt(10000));
????????????master.submit(task);
????????}

????????master.execute();

????????while?(true)?{
????????????if?(master.isComplete())?{
????????????????System.out.println("執(zhí)行的結(jié)果為:?"?+?master.getResult());
????????????????break;
????????????}
????????}

ForkJoin線程池

該線程池是jdk7之后引入的一個(gè)并行執(zhí)行任務(wù)的框架, 其核心思想也是將任務(wù)分割為子任務(wù),

有可能子任務(wù)還是很大, 還需要進(jìn)一步拆解, 最終得到足夠小的任務(wù).

將分割出來的子任務(wù)放入雙端隊(duì)列中, 然后幾個(gè)啟動(dòng)線程從雙端隊(duì)列中獲取任務(wù)執(zhí)行.

子任務(wù)執(zhí)行的結(jié)果放到一個(gè)隊(duì)列里, 另起線程從隊(duì)列中獲取數(shù)據(jù), 合并結(jié)果.

假設(shè)我們的場景需要計(jì)算從0到20000000L的累加求和. CountTask繼承自RecursiveTask, 可以攜帶返回值.

每次分解大任務(wù), 簡單的將任務(wù)劃分為100個(gè)等規(guī)模的小任務(wù), 并使用fork()提交子任務(wù).

在子任務(wù)中通過THRESHOLD設(shè)置子任務(wù)分解的閾值, 如果當(dāng)前需要求和的總數(shù)大于THRESHOLD, 則子任務(wù)需要再次分解,如果子任務(wù)可以直接執(zhí)行, 則進(jìn)行求和操作, 返回結(jié)果. 最終等待所有的子任務(wù)執(zhí)行完畢, 對(duì)所有結(jié)果求和.

public?class?CountTask?extends?RecursiveTask<Long>{
????//?任務(wù)分解的閾值
????private?static?final?int?THRESHOLD?=?10000;
????private?long?start;
????private?long?end;


????public?CountTask(long?start,?long?end)?{
????????this.start?=?start;
????????this.end?=?end;
????}

????public?Long?compute()?{
????????long?sum?=?0;
????????boolean?canCompute?=?(end?-?start)?????????if?(canCompute)?{
????????????for?(long?i?=?start;?i?<=?end;?i++)?{
????????????????sum?+=?i;
????????????}
????????}?else?{
????????????//?分成100個(gè)小任務(wù)
????????????long?step?=?(start?+?end)?/?100;
????????????ArrayList?subTasks?=?new?ArrayList();
????????????long?pos?=?start;
????????????for?(int?i?=?0;?i?100;?i++)?{
????????????????long?lastOne?=?pos?+?step;
????????????????if?(lastOne?>?end)?{
????????????????????lastOne?=?end;
????????????????}
????????????????CountTask?subTask?=?new?CountTask(pos,?lastOne);
????????????????pos?+=?step?+?1;
????????????????//?將子任務(wù)推向線程池
????????????????subTasks.add(subTask);
????????????????subTask.fork();
????????????}

????????????for?(CountTask?task?:?subTasks)?{
????????????????//?對(duì)結(jié)果進(jìn)行join
????????????????sum?+=?task.join();
????????????}
????????}
????????return?sum;
????}

????public?static?void?main(String[]?args)?throws?ExecutionException,?InterruptedException?{
????????ForkJoinPool?pool?=?new?ForkJoinPool();
????????//?累加求和?0?->?20000000L
????????CountTask?task?=?new?CountTask(0,?20000000L);
????????ForkJoinTask?result?=?pool.submit(task);
????????System.out.println("sum?result?:?"?+?result.get());
????}
}

ForkJoin線程池使用一個(gè)無鎖的棧來管理空閑線程, 如果一個(gè)工作線程暫時(shí)取不到可用的任務(wù), 則可能被掛起.

掛起的線程將被壓入由線程池維護(hù)的棧中, 待將來有任務(wù)可用時(shí), 再從棧中喚醒這些線程.

-?END -
最近熱文
? ?武大94年博士年薪201萬入職華為！學(xué)霸日程表曝光，簡直降維打擊！
? ?我在美團(tuán)的八年
? ?為什么下載小電影時(shí)，經(jīng)常會(huì)卡在99%？
? ?朝陽群眾盯上了望京A座？舉報(bào)996造成交通嚴(yán)重堵塞