多執行緒併發執行任務,取結果歸集:Future、FutureTask、CompletionService、CompletableFuture
阿新 • • 發佈:2019-02-11
Future
(1)cancle
(2)get
(3)isCancle
(4)isDone
示例:
使用執行緒池提交Callable介面任務,返回Future介面,新增進李斯特,最後遍歷FutureList且內部使用while輪詢,併發獲取結果
package thread.future; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; /** * * @ClassName: FutureDemo * @Description: Future多執行緒併發任務結果歸集 * @author denny.zhang * @date 2016年11月4日 下午1:50:32 * */ public class FutureDemo{ public static void main(String[] args) { Long start = System.currentTimeMillis(); //開啟多執行緒 ExecutorService exs = Executors.newFixedThreadPool(10); try { //結果集 List<Integer> list = new ArrayList<Integer>(); List<Future<Integer>> futureList = new ArrayList<Future<Integer>>(); //1.高速提交10個任務,每個任務返回一個Future入list for(int i=0;i<10;i++){ futureList.add(exs.submit(new CallableTask(i+1))); } Long getResultStart = System.currentTimeMillis(); System.out.println("結果歸集開始時間="+new Date()); //2.結果歸集,遍歷futureList,高速輪詢(模擬實現了併發)獲取future狀態成功完成後獲取結果,退出當前迴圈 for (Future<Integer> future : futureList) { while (true) {//CPU高速輪詢:每個future都併發輪循,判斷完成狀態然後獲取結果,這一行,是本實現方案的精髓所在。即有10個future在高速輪詢,完成一個future的獲取結果,就關閉一個輪詢 if (future.isDone()&& !future.isCancelled()) {//獲取future成功完成狀態,如果想要限制每個任務的超時時間,取消本行的狀態判斷+future.get(1000*1, TimeUnit.MILLISECONDS)+catch超時異常使用即可。 Integer i = future.get();//獲取結果 System.out.println("任務i="+i+"獲取完成!"+new Date()); list.add(i); break;//當前future獲取結果完畢,跳出while } else { Thread.sleep(1);//每次輪詢休息1毫秒(CPU納秒級),避免CPU高速輪循耗空CPU---》新手別忘記這個 } } } System.out.println("list="+list); System.out.println("總耗時="+(System.currentTimeMillis()-start)+",取結果歸集耗時="+(System.currentTimeMillis()-getResultStart)); } catch (Exception e) { e.printStackTrace(); } finally { exs.shutdown(); } } static class CallableTask implements Callable<Integer>{ Integer i; public CallableTask(Integer i) { super(); this.i=i; } @Override public Integer call() throws Exception { if(i==1){ Thread.sleep(3000);//任務1耗時3秒 }else if(i==5){ Thread.sleep(5000);//任務5耗時5秒 }else{ Thread.sleep(1000);//其它任務耗時1秒 } System.out.println("task執行緒:"+Thread.currentThread().getName()+"任務i="+i+",完成!"); return i; } } }
FutureTask
1.Runnable介面,可開啟單個執行緒執行。
2.Future<v>介面,可接受Callable介面的返回值,futureTask.get阻塞獲取結果。
/** * * @ClassName:FutureTaskDemo * @Description:FutureTask彌補了Future必須用執行緒池提交返回Future的缺陷,實現功能如下: * 1.Runnable介面,可開啟執行緒執行。 * 2.Future<v>介面,可接受Callable介面的返回值,futureTask.get()阻塞獲取結果。 * 這兩個步驟:一個開啟執行緒執行任務,一個阻塞等待執行結果,分離這兩步驟,可在這兩步中間穿插別的相關業務邏輯。 * @author diandian.zhang * @date 2017年6月16日上午10:36:05 */ public class FutureTaskContorlDemo { public static void main(String[] args) { try { System.out.println("=====例如一個統計公司總部和分部的總利潤是否達標100萬=========="); //利潤 Integer count = 0; //1.定義一個futureTask,假設去遠端http獲取各個分公司業績. FutureTask<Integer> futureTask = new FutureTask<Integer>(new CallableTask()); Thread futureTaskThread = new Thread(futureTask); futureTaskThread.start(); System.out.println("futureTaskThread start!"+new Date()); //2.主執行緒先做點別的事 System.out.println("主執行緒查詢總部公司利潤開始時間:"+new Date()); Thread.sleep(5000); count+=10;//北京集團總部利潤。 System.out.println("主執行緒查詢總部公司利潤結果時間:"+new Date()); //總部已達標100萬利潤,就不再繼續執行獲取分公司業績任務了 if(count>=100){ System.out.println("總部公司利潤達標,取消futureTask!"+new Date()); futureTask.cancel(true);//不需要再去獲取結果,那麼直接取消即可 }else{ System.out.println("總部公司利潤未達標,進入阻塞查詢分公司利潤!"+new Date()); //3總部未達標.阻塞獲取,各個分公司結果 Integer i = futureTask.get();//真正執行CallableTask System.out.println("i="+i+"獲取到結果!"+new Date()+new Date()); } } catch (Exception e) { e.printStackTrace(); } } /** * * @ClassName:CallableTask * @Description:一個十分耗時的任務 * @author diandian.zhang * @date 2017年6月16日上午10:39:04 */ static class CallableTask implements Callable<Integer>{ @Override public Integer call() throws Exception { System.out.println("CallableTask-call,查詢分公司利潤,執行開始!"+new Date()); Thread.sleep(10000); System.out.println("CallableTask-call,查詢分公司利潤,執行完畢!"+new Date()); return 10; } }
package thread.future; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.FutureTask; /** * * @ClassName:FutureTaskDemo * @Description:FutureTask實現多執行緒併發執行任務並取結果歸集 * @author diandian.zhang * @date 2017年6月16日上午10:36:05 */ public class FutureTaskDemo { public static void main(String[] args) { Long start = System.currentTimeMillis(); //開啟多執行緒 ExecutorService exs = Executors.newFixedThreadPool(5); try { //結果集 List<Integer> list = new ArrayList<Integer>(); List<FutureTask<Integer>> futureList = new ArrayList<FutureTask<Integer>>(); //啟動執行緒池,10個任務固定執行緒數為5 for(int i=0;i<10;i++){ FutureTask<Integer> futureTask = new FutureTask<Integer>(new CallableTask(i+1)); //提交任務,新增返回 exs.submit(futureTask);//Runnable特性 futureList.add(futureTask);//Future特性 } Long getResultStart = System.currentTimeMillis(); System.out.println("結果歸集開始時間="+new Date()); //結果歸集 for (FutureTask<Integer> future : futureList) { while (true) { if (future.isDone()&& !future.isCancelled()) { Integer i = future.get();//Future特性 System.out.println("i="+i+"獲取到結果!"+new Date()); list.add(i); break; }else { Thread.sleep(1);//避免CPU高速輪循,可以休息一下。 } } } System.out.println("list="+list); System.out.println("總耗時="+(System.currentTimeMillis()-start)+",取結果歸集耗時="+(System.currentTimeMillis()-getResultStart)); } catch (Exception e) { e.printStackTrace(); } finally { exs.shutdown(); } } static class CallableTask implements Callable<Integer>{ Integer i; public CallableTask(Integer i) { super(); this.i=i; } @Override public Integer call() throws Exception { if(i==1){ Thread.sleep(3000);//任務1耗時3秒 }else if(i==5){ Thread.sleep(5000);//任務5耗時5秒 }else{ Thread.sleep(1000);//其它任務耗時1秒 } System.out.println("執行緒:["+Thread.currentThread().getName()+"]任務i="+i+",完成!"+new Date()); return i; } } }
CompletionService
原因:內部通過阻塞佇列+FutureTask,實現了任務先完成可優先獲取到,即結果完成後順序排序。
package thread.future;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
*
* @ClassName: CompletionServiceDemo
* @Description: CompletionService多執行緒併發任務結果歸集
* @author denny.zhang
* @date 2016年11月4日 下午1:50:32
*
*/
public class CompletionServiceDemo{
public static void main(String[] args) {
Long start = System.currentTimeMillis();
//開啟3個執行緒
ExecutorService exs = Executors.newFixedThreadPool(5);
try {
int taskCount = 10;
//結果集
List<Integer> list = new ArrayList<Integer>();
//1.定義CompletionService
CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(exs);
List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
//2.新增任務
for(int i=0;i<taskCount;i++){
futureList.add(completionService.submit(new Task(i+1)));
}
//==================結果歸集===================
//方法1:future是提交時返回的,遍歷queue則按照任務提交順序,獲取結果
// for (Future<Integer> future : futureList) {
// System.out.println("====================");
// Integer result = future.get();//執行緒在這裡阻塞等待該任務執行完畢,按照
// System.out.println("任務result="+result+"獲取到結果!"+new Date());
// list.add(result);
// }
// //方法2.使用內部阻塞佇列的take()
for(int i=0;i<taskCount;i++){
Integer result = completionService.take().get();//採用completionService.take(),內部維護阻塞佇列,任務先完成的先獲取到
System.out.println("任務i=="+result+"完成!"+new Date());
list.add(result);
}
System.out.println("list="+list);
System.out.println("總耗時="+(System.currentTimeMillis()-start));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();//關閉執行緒池
}
}
static class Task implements Callable<Integer>{
Integer i;
public Task(Integer i) {
super();
this.i=i;
}
@Override
public Integer call() throws Exception {
if(i==5){
Thread.sleep(5000);
}else{
Thread.sleep(1000);
}
System.out.println("執行緒:"+Thread.currentThread().getName()+"任務i="+i+",執行完成!");
return i;
}
}
}
CompletableFuture
JDK1.8才新加入的實現類,實現Future、CompletionStage兩個介面。
package thread.future;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.google.common.collect.Lists;
/**
*
* @ClassName:CompletableFutureDemo
* @Description:多執行緒併發任務,取結果歸集
* @author diandian.zhang
* @date 2017年6月14日下午12:44:01
*/
public class CompletableFutureDemo {
public static void main(String[] args) {
Long start = System.currentTimeMillis();
//結果集
List<String> list = new ArrayList<String>();
List<String> list2 = new ArrayList<String>();
//定長10執行緒池
ExecutorService exs = Executors.newFixedThreadPool(10);
//List<CompletableFuture<String>> futureList = new ArrayList<>();
List<Integer> taskList = Lists.newArrayList(2,1,3,4,5,6,7,8,9,10);
try {
//方式一:迴圈建立CompletableFuture list,呼叫sequence()組裝返回一個有返回值的CompletableFuture,返回結果get()獲取
// for(int i=0;i<taskList.size();i++){
// final int j=i+1;
// CompletableFuture<String> future = CompletableFuture.supplyAsync(()->calc(j), exs)//非同步執行
// .thenApply(e->Integer.toString(e))//Integer轉換字串 thenAccept只接受不返回不影響結果
//// .whenComplete((v, e) -> {//如需獲取任務完成先手順序,此處程式碼即可
//// System.out.println("任務"+v+"完成!result="+v+",異常 e="+e+","+new Date());
//// list2.add(v);
//// })
// ;
// futureList.add(future);
// }
// //流式獲取結果
// list = sequence(futureList).get();//[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]此處不理解為什麼是這樣的順序?誰知道求告知
//方式二:全流式處理轉換成CompletableFuture[]+組裝成一個無返回值CompletableFuture,join等待執行完畢。返回結果whenComplete獲取
@SuppressWarnings("rawtypes")
CompletableFuture[] cfs = taskList.stream().map(object-> CompletableFuture.supplyAsync(()->calc(object), exs)
.thenApply(h->Integer.toString(h))
.whenComplete((v, e) -> {//如需獲取任務完成先手順序,此處程式碼即可
System.out.println("任務"+v+"完成!result="+v+",異常 e="+e+","+new Date());
list2.add(v);
}))
.toArray(CompletableFuture[]::new);
CompletableFuture.allOf(cfs).join();//封裝後無返回值,必須自己whenComplete()獲取
System.out.println("list2="+list2+"list="+list+",耗時="+(System.currentTimeMillis()-start));
} catch (Exception e) {
e.printStackTrace();
}finally {
exs.shutdown();
}
}
public static Integer calc(Integer i){
try {
if(i==1){
Thread.sleep(3000);//任務1耗時3秒
}else if(i==5){
Thread.sleep(5000);//任務5耗時5秒
}else{
Thread.sleep(1000);//其它任務耗時1秒
}
System.out.println("task執行緒:"+Thread.currentThread().getName()+"任務i="+i+",完成!+"+new Date());
} catch (InterruptedException e) {
e.printStackTrace();
}
return i;
}
/**
*
* @Description 組合多個CompletableFuture為一個CompletableFuture,所有子任務全部完成,組合後的任務才會完成。帶返回值,可直接get.
* @param futures List
* @return
* @author diandian.zhang
* @date 2017年6月19日下午3:01:09
* @since JDK1.8
*/
public static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
//1.構造一個空CompletableFuture,子任務數為入參任務list size
CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
//2.流式(每個子任務join操作後轉換為list)往空CompletableFuture中新增結果
return allDoneFuture.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.<T>toList()));
}
/**
*
* @Description Stream流式型別futures轉換成一個CompletableFuture,所有子任務全部完成,組合後的任務才會完成。帶返回值,可直接get.
* @param futures Stream
* @return
* @author diandian.zhang
* @date 2017年6月19日下午6:23:40
* @since JDK1.8
*/
public static <T> CompletableFuture<List<T>> sequence(Stream<CompletableFuture<T>> futures) {
List<CompletableFuture<T>> futureList = futures.filter(f -> f != null).collect(Collectors.toList());
return sequence(futureList);
}
}
最後歡迎大家訪問我的個人網站:1024s