Future之allAsList与successfulAsList

Future之allAsList与successfulAsList

原创 约定291天后 最后发布于2018-03-30 10:45:36 阅读数 1610 收藏

发布于2018-03-30 10:45:36

分类专栏： Java基础

版权声明：本文为博主原创文章，遵循 CC 4.0 BY 版权协议，转载请附上原文出处链接和本声明。

本文链接：https://blog.csdn.net/sunquan291/article/details/79748086

展开

Guava中Futures

1. transform：对于ListenableFuture的返回值进行转换。
2. allAsList：对多个ListenableFuture的合并，返回一个当所有Future成功时返回多个Future返回值组成的List对象。注：当其中一个Future失败或者取消的时候，将会进入失败或者取消。
3. successfulAsList：和allAsList相似，唯一差别是对于失败或取消的Future返回值用null代替。不会进入失败或者取消流程。
4. immediateFuture/immediateCancelledFuture： 立即返回一个待返回值的ListenableFuture。
5. makeChecked: 将ListenableFuture 转换成CheckedFuture。CheckedFuture 是一个ListenableFuture ，其中包含了多个版本的get 方法，方法声明抛出检查异常.这样使得创建一个在执行逻辑中可以抛出异常的Future更加容易
6. JdkFutureAdapters.listenInPoolThread(future): guava同时提供了将JDK Future转换为ListenableFuture的接口函数。
7. addCallBack为Future增加回调

package com.zte.sunquan.demo.future;
import akka.actor.AbstractActor;
import akka.actor.Props;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
/**
 * Created by sunquan on 2017/9/30.
 */
public class ReturnActor extends AbstractActor {
    @GuardedBy("DATE_FORMAT")
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
    public static Props props() {
        return Props.create(ReturnActor.class);
    }
    @Override
    public Receive createReceive() {
        return receiveBuilder().match(String.class, s -> {
            System.out.println("Got String:" + s + ", at:" + DATE_FORMAT.format(new Date()));
            getContext().stop(getSelf());
            getContext().actorOf(ReturnActor.props(),"print");
//            TimeUnit.SECONDS.sleep(5);
//            getSender().tell(28,getSelf());
        }).build();
    }
}

package com.zte.sunquan.demo.future;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.dispatch.OnComplete;
import akka.pattern.Patterns;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import javax.annotation.Nullable;
import org.junit.Before;
import org.junit.Test;
import scala.concurrent.Future;
/**
 * Created by sunquan on 2018/3/28.
 */
public class FutureAsListTest {
    ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
    ListenableFuture future1 = null;
    ListenableFuture future2 = null;
    static ActorSystem system = ActorSystem.create("mysystem");
    @Before
    public void init() {
        future1 = service.submit(new Callable<Integer>() {
            public Integer call() throws InterruptedException {
                Thread.sleep(1000);
                System.out.println("call future 1.");
                return 1;
            }
        });
        future2 = service.submit(new Callable<Integer>() {
            public Integer call() throws InterruptedException {
                Thread.sleep(1000);
                System.out.println("call future 2.");
                //       throw new RuntimeException("----call future 2.");
                return 2;
            }
        });
    }
    @Test
    public void testAllAsList() throws Exception {
        //allAsList
        final ListenableFuture<List<Integer>> allFutures = Futures.allAsList(future1, future2);
        Futures.addCallback(allFutures, new FutureCallback<List<Integer>>() {
            @Override
            public void onSuccess(@Nullable List<Integer> result) {
                result.forEach(p -> {
                    System.out.println("output: " + p);
                });
            }
            @Override
            public void onFailure(Throwable t) {
                System.out.println("Output error " + t);
            }
        });
        System.out.println(allFutures.get());
        Thread.sleep(2000);
//        call future 2.
//        call future 1.
//        [1, 2]
//        Output: 1
//        Output: 2
    }
    @Test
    public void testAllAsListWithException() throws Exception {
        future2.cancel(true);
        ListenableFuture future3 = service.submit(new Callable<Integer>() {
            public Integer call() throws InterruptedException {
                Thread.sleep(1000);
                System.out.println("call future 3.");
                throw new RuntimeException("----call future 3 error.");
            }
        });
        //allAsList
        final ListenableFuture<List<Integer>> allFutures = Futures.allAsList(future1, future3);
        Futures.addCallback(allFutures, new FutureCallback<List<Integer>>() {
            @Override
            public void onSuccess(@Nullable List<Integer> result) {
                for (Integer r : result)
                    System.out.println("Output: " + r);
            }
            @Override
            public void onFailure(Throwable t) {
                System.out.println("Output error " + t);
            }
        });
        System.out.println(allFutures.get());
//        call future 1.
//        call future 3.
//        Output error java.lang.RuntimeException: ----call future 3 error.
//        future3抛出异常，针对allAsList，直接走Failure流程，不管其它feture有没有正确完成
//        java.util.concurrent.ExecutionException: java.lang.RuntimeException: ----call future 3 errror.
    }
    @Test
    public void testAllAsListWithException2() throws Exception {
        future2.cancel(true);
        final SettableFuture<Object> settableFuture = SettableFuture.create();
        ActorRef print = system.actorOf(ReturnActor.props(), "print");
        //注意此处2秒超时，如果这个ask超时，下面的聚合后，还是会走onFailure
        //什么时候，会中断？
        Future<Object> ask = Patterns.ask(print, "how old are you", 2000);
        ask.onComplete(new OnComplete<Object>() {
            @Override
            public void onComplete(Throwable failure, Object o) throws Throwable {
                if (failure != null) {
                    settableFuture.setException(failure);
                } else
                    settableFuture.set(o);
            }
        }, system.dispatcher());
        //allAsList
        final ListenableFuture<List<Integer>> allFutures = Futures.allAsList(future1, settableFuture);
        Futures.addCallback(allFutures, new FutureCallback<List<Integer>>() {
            @Override
            public void onSuccess(@Nullable List<Integer> result) {
                for (Integer r : result)
                    System.out.println("Output: " + r);
            }
            @Override
            public void onFailure(Throwable t) {
                System.out.println("Output error " + t);
            }
        });
        System.out.println(allFutures.get());
    }
    @Test
    public void testSuccessfulAsListWithException() throws Exception {
        future2.cancel(true);
        ListenableFuture future4 = service.submit(new Callable<Integer>() {
            public Integer call() throws InterruptedException {
                Thread.sleep(1000);
                System.out.println("call future 4.");
                throw new RuntimeException("----call future 4.");
            }
        });
        //successfulAsList
        final ListenableFuture<List<Integer>> allFutures = Futures.successfulAsList(future1, future4);
        Futures.addCallback(allFutures, new FutureCallback<List<Integer>>() {
            @Override
            public void onSuccess(@Nullable List<Integer> result) {
                for (Integer r : result)
                    System.out.println("Output: " + r);
            }
            @Override
            public void onFailure(Throwable t) {
                System.out.println("Output error " + t);
            }
        });
        System.out.println(allFutures.get());
//        call future 1.
//        call future 4.
//        对于future4抛出异常，其返回值为null,聚合Future不会走failure流程
//        [1, null]
//        Output: 1
//        Output: null
    }
    @Test
    public void should_test_furture() throws Exception {
        //allAsList
        final ListenableFuture<List<Integer>> allFutures = Futures.allAsList(future1, future2);
        Futures.addCallback(allFutures, new FutureCallback<List<Integer>>() {
            @Override
            public void onSuccess(@Nullable List<Integer> result) {
                for (Integer r : result)
                    System.out.println("Output: " + r);
            }
            @Override
            public void onFailure(Throwable t) {
                System.out.println("Output error " + t);
            }
        });
        //Futures transform可以进一步重定义结果
        final ListenableFuture transform = Futures.transform(allFutures, new AsyncFunction<List<Integer>, Boolean>() {
            @Override
            public ListenableFuture apply(List<Integer> results) throws Exception {
                if (results.size() == 2)
                    return Futures.immediateFuture(true);
                else
                    return Futures.immediateFuture(false);
            }
        });
        Futures.addCallback(transform, new FutureCallback<Object>() {
            public void onSuccess(Object result) {
                System.out.println("success with: " + result);
            }
            public void onFailure(Throwable thrown) {
                System.out.printf("onFailure%s\n", thrown.getMessage());
            }
        });
        System.out.println(transform.get());
//        call future 1.
//        call future 2.
//        Output: 1
//        Output: 2
//        true
//        success with: true
    }
}