Java的Future模式

Future接口的作用

Future表示异步计算的结果。 提供了一些方法来检查计算是否完成，等待其完成以及检索计算结果。 只有在计算完成时才可以使用get方法检索结果，必要时将其阻塞，直到准备就绪为止。 取消由cancel方法执行。 提供了其他方法来确定任务是正常完成还是被取消。 一旦计算完成，就不能取消计算。 如果出于可取消性的目的而使用Future而不提供可用的结果，则可以声明Future <?>形式的类型，并作为基础任务的结果返回null。

Future接口提供的方法

    // 取消任务
    boolean cancel(boolean mayInterruptIfRunning);
    // 任务是否被取消
    boolean isCancelled();
    // 任务是否完成
    boolean isDone();
    // 返回Future所指向的异步任务的执行结果
    // 当get方法被调用的时候，异步的任务还没有完成，get将进入阻塞状态
    V get() throws InterruptedException, ExecutionException;
    // 如果超时时间到了，异步任务还没有完成，throws TimeoutException
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

实现类FutureTask

public class FutureTask<V> implements RunnableFuture<V> {}
public interface RunnableFuture<V> extends Runnable, Future<V> {}

	// 有返回值
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    // 无返回值
    public FutureTask(Runnable runnable, V result) {
        // 将Runnable 转成Callable
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

package com.learn.thread.future;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class FutureTest {

    public static void main(String[] args) {
        Callable<Integer> callable = () -> {
            System.out.println("pre execution");

            int randomNumber = new Random().nextInt(500);

            System.out.println("post execution");

            return randomNumber;
        };

        FutureTask futureTask = new FutureTask(callable);
        new Thread(futureTask).start();

        System.out.println("thread has started");
        try {
            System.out.println(futureTask.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

package com.learn.thread.future;

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

public class FutureTest {

    public static void main(String[] args) {
        Callable<Integer> callable = () -> {
            System.out.println("pre execution");

            Thread.sleep(5000);

            int randomNumber = new Random().nextInt(500);

            System.out.println("post execution");

            return randomNumber;
        };

        FutureTask futureTask = new FutureTask(callable);
        new Thread(futureTask).start();

        System.out.println("thread has started");
        try {
            Thread.sleep(2000);
            System.out.println(futureTask.get(1, TimeUnit.MILLISECONDS));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

主线程抛出异常，异步任务不受影响。

CompletableFuture

弥补Future的不足（get方法阻塞）
一个可以被显式执行的Future（设置它的值和状态），还可以当做一个CompletionStage来使用，它支持在完成时触发的相关功能和操作。
当有两个或者多个线程去尝试complete，completeExceptionally或者cancel一个CompletableFuture时，只有一个会成功。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {}

package com.learn.thread.future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class CmpFutureTest {

    public static void main(String[] args) {
        // 阶段性完成
        // 对结果进行转换
        String result = CompletableFuture.supplyAsync(() -> "hello")
                .thenApplyAsync(value -> value + " world").join();
        System.out.println(result);

        // 对结果进行消费
        CompletableFuture.supplyAsync(() -> "hello").thenAccept(value -> System.out.println("welcome " + value));

        // 对于storage的合并
        String result2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "world";
        }), (s1, s2) -> s1 + " " + s2).join();

        System.out.println(result2);
    }
}

package com.learn.thread.future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class CmpFutureTest {

    public static void main(String[] args) {
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task finished");
        });
		// 主线程不阻塞
        completableFuture.whenComplete((t, action) -> System.out.println("执行完成!"));

        System.out.println("主线程执行完毕！");

        try {
            TimeUnit.MILLISECONDS.sleep(7000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

本文地址：https://blog.csdn.net/wz122330/article/details/107662021