Java多線程的Callable, Future, FutureCallback

Callable能夠當作是一個加強版的Runnable, 帶返回結果, 須要經過Future或者FutureTask來提交任務或運行線程, 而後經過Future/FutureTask的get方法獲得返回結果.

Callable在子線程中運行, 在主線程中異步獲得執行結果(get()方法是阻塞的), 或者檢查是否已取消, 是否已完成(檢查取消和完成的方法是非阻塞的)

經過Thread子線程啓動

這種方式, 須要建立一個FutureTask對象, 再用這個FutureTask對象建立一個Thread來運行. 後續操做都經過FutureTask進行.

public class DemoCallableFuture {
    public static void main(String[] args) {
        FutureTask<String> task = new FutureTask<>(()->{
            System.out.println("task start");
            Thread.sleep(1000);
            System.out.println("task done");
            return "task get";
        });

        new Thread(task).start();

        FutureTask<String> task2 = new FutureTask<>(()->{
            System.out.println("task2 start");
            Thread.sleep(1000);
            System.out.println("task2 done");
            return "task2 get";
        });

        new Thread(task2).start();

        if (task.isCancelled()) {
            System.out.println("task cancelled yes");
        } else {
            System.out.println("task cancelled no");
        }
        if (task.isDone()) {
            System.out.println("task done yes");
        } else {
            System.out.println("task done no");
        }

        try {
            System.out.println(task.get());
        } catch (InterruptedException|ExecutionException e) {
            e.printStackTrace();
        }

        if (task2.isCancelled()) {
            System.out.println("task2 cancelled yes");
        } else {
            System.out.println("task2 cancelled no");
        }
        if (task2.isDone()) {
            System.out.println("task2 done yes");
        } else {
            System.out.println("task2 done no");
        }
        try {
            System.out.println(task2.get());
        } catch (InterruptedException|ExecutionException e) {
            e.printStackTrace();
        }
    }
}

運行結果

task start
task cancelled no
task done no
task2 start
task2 done
task done
task get
task2 cancelled no
task2 done yes
task2 get

　　

經過ExecutorService線程池啓動

這種方式, 經過線程池submit一個Callable對象, 就會獲得一個Future對象, 根據這個Future對象作後續操做

public class DemoCallableFuture2 {
    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(4);
        Future<String> future = service.submit(()->{
            System.out.println("task start");
            Thread.sleep(1000);
            System.out.println("task done");
            return "task get";
        });

        if (future.isCancelled()) {
            System.out.println("task cancelled yes");
        } else {
            System.out.println("task cancelled no");
        }
        if (future.isDone()) {
            System.out.println("task done yes");
        } else {
            System.out.println("task done no");
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        if (future.isCancelled()) {
            System.out.println("task cancelled yes");
        } else {
            System.out.println("task cancelled no");
        }
        if (future.isDone()) {
            System.out.println("task done yes");
        } else {
            System.out.println("task done no");
        }

        try {
            System.out.println(future.get());
        } catch (InterruptedException|ExecutionException e) {
            e.printStackTrace();
        }
    }
}

運行結果

task cancelled no
task done no
task start
task cancelled no
task done no
task done
task get

.

FutureCallback

FutureCallback是Google Guava中的一個類, 解決的是Future中get阻塞的問題, 讓全過程異步. 須要使用ListeningExecutorService的線程池提交.

代碼例子

public class DemoFutureCallback {
    public static void main(String[] args) {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));

        for (int i = 0; i < 5; i++) {
            int j = i;
            ListenableFuture<String> future = service.submit(()->{
                System.out.println("task start");
                Thread.sleep(1000);
                System.out.println("task done");
                return "task return " + j;
            });

            Futures.addCallback(future, new FutureCallback<String>() {
                @Override
                public void onSuccess(String s) {
                    System.out.println("callback success: " + s);
                }

                @Override
                public void onFailure(Throwable throwable) {
                    throwable.printStackTrace();
                }
            });
        }
        System.out.println("thread ongoing");
        service.shutdown();
    }
}

運行結果

task start
task start
task start
thread ongoing
task start
task start
task done
callback success: task return 0
task done
callback success: task return 1
task done
callback success: task return 3
task done
callback success: task return 4
task done
callback success: task return 2

Process finished with exit code 0