当前位置：首页 > news >正文

Future模式-多任务并行汇总 or 同步任务回调

news 来源：原创 2024/4/19 11:40:22

2019独角兽企业重金招聘Python工程师标准>>>

多任务并行汇总 or 同步任务回调

使用JDK提供的类和方法来演示，方式一异步多任务并行汇总：

import java.util.concurrent.Callable;

public class RealData implements Callable<String> {

private String para;

public RealData(String para) {

this.para = para;

}

@Override

public String call() throws Exception {

StringBuffer sb = new StringBuffer();

for (int i = 0; i < 10; i++) {

sb.append(para);

try {

Thread.sleep(100);

} catch (InterruptedException e) {

}

}

return sb.toString();

}

}

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.FutureTask;

public class FutureMain {

public static void main(String[] args) throws InterruptedException,

ExecutionException {

// 构造FutureTask

FutureTask<String> future = new FutureTask<String>(new RealData("a"));

ExecutorService executor = Executors.newFixedThreadPool(1);

// 执行FutureTask，相当于上例中的 client.request("a") 发送请求

// 在这里开启线程进行RealData的call()执行

executor.submit(future);

System.out.println("请求完毕");

try {

// 这里依然可以做额外的数据操作，这里使用sleep代替其他业务逻辑的处理

Thread.sleep(2000);

} catch (InterruptedException e) {

}

// 相当于data.getResult ()，取得call()方法的返回值

// 如果此时call()方法没有执行完成，则依然会等待

System.out.println("数据 = " + future.get());

}

}

这里要注意的是FutureTask是即具有 Future功能又具有Runnable功能的类。所以又可以运行，最后还能get。

当然如果在调用到future.get()时，真实数据还没准备好，仍然会产生阻塞状况，直到数据准备完成。

当然还有更加简便的方式二同步任务回调：

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class FutureMain2 {

public static void main(String[] args) throws InterruptedException,

ExecutionException {

ExecutorService executor = Executors.newFixedThreadPool(1);

// 执行FutureTask，相当于上例中的 client.request("a") 发送请求

// 在这里开启线程进行RealData的call()执行

Future<String> future = executor.submit(new RealData("a"));

System.out.println("请求完毕");

try {

// 这里依然可以做额外的数据操作，这里使用sleep代替其他业务逻辑的处理

Thread.sleep(2000);

} catch (InterruptedException e) {

}

// 相当于data.getResult ()，取得call()方法的返回值

// 如果此时call()方法没有执行完成，则依然会等待

System.out.println("数据 = " + future.get());

}

}

由于Callable是有返回值的，可以直接返回future对象。

---------------------------------------------------------------------
JDK8提供了CompletionStage接口（40余个方法），大多数方法多数应用在函数式编程中，并且支持流式调用。是Java 8中对Future的增强版。

简单实现：

import java.util.concurrent.CompletableFuture;

public class AskThread implements Runnable {

CompletableFuture<Integer> re = null;

public AskThread(CompletableFuture<Integer> re) {

this.re = re;

}

@Override

public void run() {

int myRe = 0;

try {

myRe = re.get() * re.get();

} catch (Exception e) {

}

System.out.println(myRe);

}

public static void main(String[] args) throws InterruptedException {

final CompletableFuture<Integer> future = new CompletableFuture<Integer>();

new Thread(new AskThread(future)).start();

// 模拟长时间的计算过程

Thread.sleep(1000);

// 告知完成结果

future.complete(60);

}

}

Future最令人诟病的就是要等待，要自己去检查任务是否完成了，在Future中，任务完成的时间是不可控的。而CompletableFuture的最大改进在于，任务完成的时间也开放了出来。

future.complete(60);

用来设置完成时间。

CompletableFuture的异步执行：

public static Integer calc(Integer para) {

try {

// 模拟一个长时间的执行

Thread.sleep(1000);

} catch (InterruptedException e) {

}

return para * para;

}

public static void main(String[] args) throws InterruptedException,

ExecutionException {

final CompletableFuture<Integer> future = CompletableFuture

.supplyAsync(() -> calc(50));

System.out.println(future.get());

}

CompletableFuture的流式调用：

public static Integer calc(Integer para) {

try {

// 模拟一个长时间的执行

Thread.sleep(1000);

} catch (InterruptedException e) {

}

return para * para;

}

public static void main(String[] args) throws InterruptedException,

ExecutionException {

CompletableFuture<Void> fu = CompletableFuture

.supplyAsync(() -> calc(50))

.thenApply((i) -> Integer.toString(i))

.thenApply((str) -> "\"" + str + "\"")

.thenAccept(System.out::println);

fu.get();

}

组合多个CompletableFuture：

public static Integer calc(Integer para) {

return para / 2;

}

public static void main(String[] args) throws InterruptedException,

ExecutionException {

CompletableFuture<Void> fu = CompletableFuture

.supplyAsync(() -> calc(50))

.thenCompose(

(i) -> CompletableFuture.supplyAsync(() -> calc(i)))

.thenApply((str) -> "\"" + str + "\"")

.thenAccept(System.out::println);

fu.get();

}