Feture、ListenableFuture、ComplatableFuture

一.Future的介绍

1.1 官方介绍

　　Future是JDK1.5中提供的一个接口（interface），关于Future，jdk文档中是这样介绍的：

Future表示“进行异步计算结果”，提供了方法来检测异步计算是否完成，以及获取计算后的结果。异步计算完后后，只能通过get方法来获取计算结果，并且使用get方法是会阻塞直到计算完毕（能拿到计算结果）。可以使用cancel方法来取消异步执行，除此之外，还提供了其他方法来判断异步任务正常完成或者被取消。一旦异步计算任务已经完成，那么任务就不能再被取消；

　　

1.2 Job和Task

　　先说一下Job和Task，最初在做项目的时候，碰到过job和task，当时并不明白有什么区别和联系，只是笼统的都成为任务。后来明白了，当时项目是每天都有一个定时任务，这个定时任务会开启多个线程去执行多个逻辑（任务），其实那个定时任务就被称为一个Job，而开启的多个线程就称为Task。需要注意的是，并不是说进程就应该称为Job，线程就应该称为Task，而是说Job和Task是“父子关系”，一个进程Job也可以开启多个进程Task。

　　而本文说的Future可以理解为Task（任务）。

　　

二. Future的使用方式

2.1 使用Callable+Future

　　两个注意点：

　　1.需要依赖实现Callable接口，注意不能是Runnable接口，因为需要拿到计算结果，所以需要实现Callable接口；

　　2.需要依赖于线程池，且要用submit提交task，而不能使用execute方法，因为execute方法提交task后，没有返回值，且task有异常也不能感知。

package cn.ganlixin;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * 描述:
 * 学习使用Future
 *
 * @author ganlixin
 * @create 2019-12-20
 */
@Slf4j
public class UseFuture {

    /**
     * 使用线程池的submit方式提交task，获取Future
     */
    @Test
    public void testExecutorSubmit() throws ExecutionException, InterruptedException {
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 提交task
        Future<String> future = executor.submit(new MyTask("hello"));

        // get方法会阻塞直到获取到结果或者捕获到异常才继续执行后面的代码
        String result = future.get();

        log.info(result);
    }

    private static class MyTask implements Callable<String> {

        private String data;

        public MyTask(String data) {
            this.data = data;
        }

        @Override
        public String call() throws Exception {
            // 模拟业务逻辑计算耗时
            TimeUnit.SECONDS.sleep(3);
            return data + " " + data;
        }
    }
}

　　

2.2 使用FutureTask+Callable

　　使用FutureTask+Callable实现时，最重要的就是拿不到返回值，即使是提供了get()方法来获取结果，拿到的结果也是null。

package cn.ganlixin;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * 描述:
 * 学习使用Future
 *
 * @author ganlixin
 * @create 2019-12-20
 */
@Slf4j
public class UseFuture {

    @Test
    public void testFutureTask() throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        FutureTask<String> futureTask = new FutureTask<>(new MyTask("hello"));
        Future<?> future = executor.submit(futureTask);

        // 阻塞，知道任务完成或者抛出异常
        Object res = future.get();

        // 预期得到的结果是"hello hello"，但是res的结果是null，因为FutureTask+Callable接口拿不到返回值
        System.out.println(res); // null
    }

    private static class MyTask implements Callable<String> {
        private String data;

        public MyTask(String data) {
            this.data = data;
        }

        @Override
        public String call() throws Exception {
            // 模拟业务逻辑计算耗时
            TimeUnit.SECONDS.sleep(3);
            log.info("MyTask.call is running, msg:{}", data);
            return data + " " + data;
        }
    }
}

　　上面使用future.get()阻塞等待结果的时候，拿到的结果是null，如果需要设定默认值，可以在submit的时候，提供第二个参数（默认返回值）。

三. ListenableFuture

　　ListenableFuture是guava提供的一种继承自JDK Future的一个接口，提供了更多的功能，比如对Future增加回调操作（callback），在成功或者失败的时候进行什么操作。

package cn.ganlixin;

import com.google.common.base.Function;
import com.google.common.util.concurrent.*;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.junit.Test;

import java.util.concurrent.*;

/**
 * 描述:
 * 学习使用Guava的ListenableFuture和ListenableFutureTask
 *
 * @author ganlixin
 * @create 2019-12-21
 */
@Slf4j
public class UseListenableFuture {

    private static class MyCallableTask implements Callable<String> {
        private String data;

        public MyCallableTask(String data) {
            this.data = data;
        }

        @Override
        public String call() throws Exception {
            // 模拟业务逻辑计算耗时
            // TimeUnit.SECONDS.sleep(3);
            log.info("MyCallableTask.call is running, msg:{}", data);
            return data + " " + data;
        }
    }

    /**
     * 创建ListenableExecutorService、ListenableFuture、FutureCallback
     */
    @Test
    public void testSimple() throws ExecutionException, InterruptedException {
        // 原始的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        // 创建ListeningExecutorService对象，使用MoreExecutors.listeningDecorator()进行包装
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);

        // 使用方式和executorService一样进行submit
        ListenableFuture<String> listenableFuture = listeningExecutorService.submit(new MyCallableTask("hello"));

        // 创建Future的回调操作，分别是成功异常的时候
        FutureCallback<String> futureCallback = new FutureCallback<String>() {
            @Override
            public void onSuccess(@Nullable String result) {
                log.info("success to run MyCallableTask, and result is:{}", result);
            }

            @Override
            public void onFailure(Throwable t) {
                log.error("fail to run MyCallableTask, e=", t);
            }
        };

        // 对Future、FutureCallBack、ExecutorService进行绑定
        // addCallback(ListenableFuture<V>, FutureCallback，Executor）
        Futures.addCallback(listenableFuture, futureCallback, listeningExecutorService);

        final String result = listenableFuture.get();
        log.info("result:{}", result);
        /*输出结果
        INFO  [pool-2-thread-1] cn.ganlixin.UseListenableFuture - MyCallableTask.call is running, msg:hello
        INFO  [main] cn.ganlixin.UseListenableFuture - result:hello hello
        INFO  [pool-2-thread-2] cn.ganlixin.UseListenableFuture - success to run MyCallableTask, and result is:hello hello
        */
    }

    /**
     * 为ListenableFuture添加Listener，Listener是在Future执行完毕后执行
     */
    @Test
    public void testAddListenr() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);
        ListenableFuture<String> listenableFuture = listeningExecutorService.submit(new MyCallableTask("hello"));

        // 可以为一个ListenableFuture添加多个Listener
        // void addListener(Runnable listener, Executor executor);
        listenableFuture.addListener(new Runnable() {
            @Override
            public void run() {
                log.info("this is listener1 running");
            }
        }, listeningExecutorService);

        listenableFuture.addListener(new Runnable() {
            @Override
            public void run() {
                log.info("this is listener2 running");
            }
        }, listeningExecutorService);

        String result = listenableFuture.get();
        log.info("result:{}", result);
    }

    /**
     * 可以使用Futures.transform进行多个Future的异步链式执行
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void testTransform() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);
        ListenableFuture<String> listenableFuture = listeningExecutorService.submit(new MyCallableTask("hello"));

        // ListenableFuture执行完毕后，将会执行下面这个Function中定义的apply方法
        ListenableFuture<String> task1 = Futures.transform(listenableFuture, new Function<String, String>() {
            @Nullable
            @Override
            public String apply(@Nullable String input) {
                // input是上一个Future的结果
                log.info("task1 input:{}", input);

                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                return "this is result from task1.apply";
            }
        }, listeningExecutorService);

        // task1执行完毕后，将会执行下面这个Function中定义的apply方法
        ListenableFuture<String> task2 = Futures.transform(task1, new Function<String, String>() {
            @Nullable
            @Override
            public String apply(@Nullable String input) {
                // input是上一个Future的结果
                log.info("task2 input:{}", input);
                return "this is result from task2.apply";
            }
        }, listeningExecutorService);

        String result = listenableFuture.get();
        log.info("result:{}", result);

        TimeUnit.SECONDS.sleep(5);
        /*
        INFO  [pool-2-thread-1] cn.ganlixin.UseListenableFuture - MyCallableTask.call is running, msg:hello
        INFO  [main] cn.ganlixin.UseListenableFuture - result:hello hello
        INFO  [pool-2-thread-2] cn.ganlixin.UseListenableFuture - task1 input:hello hello
        INFO  [pool-2-thread-3] cn.ganlixin.UseListenableFuture - task2 input:this is result from task1.apply
         */
    }

    /**
     * 使用ListenableFutureTask
     */
    @Test
    public void testListenableFutureTask() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);

        // 传入Callable接口
        ListenableFutureTask<String> listenableFutureTask = ListenableFutureTask.create(new MyCallableTask("Hello"));
        ListenableFuture<?> future = listeningExecutorService.submit(listenableFutureTask);

        // 添加Listener的方式相同
        listenableFutureTask.addListener(new Runnable() {
            @Override
            public void run() {
                log.info("this is listener of listeneFutureTask");
            }
        }, executorService);

        future.get();
        /*
        INFO  [pool-2-thread-1] cn.ganlixin.UseListenableFuture - MyCallableTask.call is running, msg:Hello
        INFO  [pool-2-thread-2] cn.ganlixin.UseListenableFuture - this is listener of listeneFutureTask
         */
    }
}

　　

四. ComplatableFuture的用法

　　CompletableFuture也是guava提供的一种异步计算的使用方式，下面简单演示了用法：

package cn.ganlixin;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 描述:
 * 学习使用ComplateableFuture
 *
 * @author ganlixin
 * @create 2019-12-22
 */
@Slf4j
public class UseCompletableFuture {

    /**
     * 使用ComplateableFuture
     */
    @Test
    public void testSupplyAsync() {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        // 可以指定第二个参数（线程池），也可以忽略，一般建议添加
        CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            log.info("this is complatebleFuture supplyAsync1 log out");
            return "task1 return";
        }, executorService);

        // 可以指定第二个参数（线程池），也可以忽略，一般建议添加
        CompletableFuture<String> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            log.info("this is complatebleFuture supplyAsync2 log out");
            return "task2 return";
        }, executorService);

        // 可以指定第二个参数（线程池），也可以忽略，一般建议添加
        CompletableFuture<String> completableFuture3 = CompletableFuture.supplyAsync(() -> {
            log.info("this is complatebleFuture supplyAsync3 log out");
            return "task3 return";
        }, executorService);

        /**
         * 输出
         * INFO  [pool-2-thread-3] cn.ganlixin.UseCompletableFuture - this is complatebleFuture supplyAsync3 log out
         * INFO  [pool-2-thread-2] cn.ganlixin.UseCompletableFuture - this is complatebleFuture supplyAsync2 log out
         * INFO  [pool-2-thread-1] cn.ganlixin.UseCompletableFuture - this is complatebleFuture supplyAsync1 log out
         */
    }

    /**
     * 异步执行链
     */
    @Test
    public void testThen() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            log.info("this is complatebleFuture supplyAsync1 log out");
            return "task1 return";
        }, executorService);

        CompletableFuture<String> completableFuture1 = completableFuture.thenApplyAsync(input -> {
            log.info("this is thenApplyAsync, input:{}", input);
            return "thenApplyAsync return";
        }, executorService);

        String result = completableFuture.get();
        log.info("result:{}", result);
        
        /*输出
        INFO  [pool-2-thread-1] cn.ganlixin.UseCompletableFuture - this is complatebleFuture supplyAsync1 log out
        INFO  [main] cn.ganlixin.UseCompletableFuture - result:task1 return
        INFO  [pool-2-thread-2] cn.ganlixin.UseCompletableFuture - this is thenApplyAsync, input:task1 return
         */
    }
}

　　

Published by

风君子

独自遨游何稽首 揭天掀地慰生平

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注