文档章节

并发编程之Callable和Future接口、FutureTask类

小菜鸡1
 小菜鸡1
发布于 2016/08/11 22:02
字数 2375
阅读 61
收藏 4

    Callable接口代表一段可以调用并返回结果的代码;Future接口表示异步任务,是还没有完成的任务给出的未来结果。所以说Callable用于产生结果,Future用于获取结果。

    Java 5在concurrency包中引入了java.util.concurrent.Callable 接口,它和Runnable接口很相似,但它可以返回一个对象或者抛出一个异常。其中Runnable可以提交给Thread来包装下,直接启动一个线程来执行,而Callable则一般都是提交给ExecuteService来执行。Executor就是Runnable和Callable的调度容器,Future就是对于具体的调度任务的执行结果进行查看,最为关键的是Future可以检查对应的任务是否已经完成,也可以阻塞在get方法上一直等待任务返回结果。Runnable和Callable的差别就是Runnable是没有结果可以返回的,并且Runnable无法抛出返回结果的异常,就算是通过Future也看不到任务调度的结果的。

    Callable接口使用泛型去定义它的返回类型。Executors类提供了一些有用的方法在线程池中执行Callable内的任务。由于 Callable任务是并行的(并行就是整体看上去是并行的,其实在某个时间点只有一个线程在执行),我们必须等待它返回的结果。

    java.util.concurrent.Future对象为我们解决了这个问题。在线程池提交Callable任务后返回了一个Future对象,使用它可以知道Callable任务的状态和得到Callable返回的执行结果。Future提供了get()方法让我们可以等待Callable结束并 获取它的执行结果。

Callable接口的源码如下:

public interface Callable<V> {
    V call() throws Exception; // 计算结果
}

Future接口的源码如下:

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);// 试图取消对此任务的执行

    boolean isCancelled();// 如果在任务正常完成前将其取消,则返回 true

    boolean isDone();// 如果任务已完成,则返回 true

    V get() throws InterruptedException, ExecutionException;// 如有必要,等待计算完成,然后获取其结果

    // 如有必要,最多等待为使计算完成所给定的时间之后,获取其结果(如果结果可用)。
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

Future用于表示异步计算的结果。它的实现类是FutureTask。

如果不想分支线程阻塞主线程,又想取得分支线程的执行结果,就用FutureTask
FutureTask实现了Runnable和Future接口,这个接口的定义如下:

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

可以看到这个接口实现了Runnable和Future接口,接口中的具体实现由FutureTask来实现。这个类的两个构造方法如下 :

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    sync = new Sync(callable);
}

public FutureTask(Runnable runnable, V result) {
    sync = new Sync(Executors.callable(runnable, result));
}

如上提供了两个构造函数,一个以Callable为参数,另外一个以Runnable为参数。这些类之间的关联对于任务建模的办法非常灵活,允许你基于 FutureTask的Runnable特性(因为它实现了Runnable接口),把任务写成Callable,然后封装进一个由执行者调度并在必要时 可以取消的FutureTask。

FutureTask可以由执行者调度,这一点很关键。它对外提供的方法基本上就是Future和Runnable接口的组合:get()、cancel、isDone()、isCancelled()和run(),而run()方法通常都是由执行者调用,我们基本上不需要直接调用它。

下面来看一个FutureTask的例子,如下:

package test1;
import java.util.concurrent.*;

class MyCallable implements Callable<String> {
    private long waitTime;

    public MyCallable(int timeInMillis) {
        this.waitTime = timeInMillis;
    }

    @Override
    public String call() throws Exception {
        Thread.sleep(waitTime);
        //return the thread name executing this callable task
        return Thread.currentThread().getName();
    }
}

public class FutureTaskExample {
    public static void main(String[] args) {
        // 要执行的任务
        MyCallable callable1 = new MyCallable(1000);
        MyCallable callable2 = new MyCallable(2000);
        // Callable写的任务封装到一个由执行者调度的FutureTask对象
        FutureTask<String> futureTask1 = new FutureTask<String>(callable1);
        FutureTask<String> futureTask2 = new FutureTask<String>(callable2);
        // 创建线程池并返回ExecutorService实例
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(futureTask1);  // 执行任务
        executor.execute(futureTask2);

        while (true) {
            try {
                //两个任务都完成
                if (futureTask1.isDone() && futureTask2.isDone()) {
                    System.out.println("Done");
                    // 关闭线程池和服务
                    executor.shutdown();
                    return;
                }
                //任务1没有完成,会等待,直到任务完成
                if (!futureTask1.isDone()) {
                    System.out.println("FutureTask1 output=" + futureTask1.get());
                }

                System.out.println("Waiting for FutureTask2 to complete");
                String s = futureTask2.get(200L, TimeUnit.MILLISECONDS);
                if (s != null) {
                    System.out.println("FutureTask2 output=" + s);
                }
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                //do nothing
            }
        }
    }
}

运行如上程序后,可以看到一段时间内没有输出,因为get()方法等待任务执行完成然后才输出内容. 

输出结果如下:

FutureTask1 output=pool-1-thread-1

Waiting for FutureTask2 to complete

Waiting for FutureTask2 to complete

Waiting for FutureTask2 to complete

Waiting for FutureTask2 to complete

Waiting for FutureTask2 to complete

FutureTask2 output=pool-1-thread-2

Done

Callable和Future接口示例程序:该程序是计算一个公司的年销售水泥的总 数目,每一行代表一个客户,每一列代表一个客户在每个月内的购买数量,将每一个客户(每一行)看做一个小任务。每一个任务计算之后放入Future中,等 待所有的计算完毕后,调用get(是一个阻塞方法,等待所有任务执行完毕)方法得到结果并计算总和。

package test1;

import java.text.DateFormatSymbols;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class AnnualSalesCalc {

    private static int NUMBER_OF_CUSTOMERS = 100;
    private static int NUMBER_OF_MONTHS = 12;
    private static int salesMatrix[][];

    private static class Summer implements Callable<Integer> {
        private int companyID;

        public Summer(int companyID) {
            this.companyID = companyID;
        }

        @Override
        public Integer call() {
            int sum = 0;
            for (int col = 0; col < NUMBER_OF_MONTHS; col++) {
                sum += salesMatrix[companyID][col];
            }
            System.out.printf("Totaling for client 1%02d completed%n",
                    companyID);
            return sum;
        }
    }

    private static void generateMatrix() {
        salesMatrix = new int[NUMBER_OF_CUSTOMERS][NUMBER_OF_MONTHS];
        for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
            for (int j = 0; j < NUMBER_OF_MONTHS; j++) {
                salesMatrix[i][j] = (int) (Math.random() * 100);
            }
        }
    }

    private static void printMatrix() {
        System.out.print("\t\t");
        String[] monthDisplayNames = (new DateFormatSymbols()).getShortMonths();
        for (String strName : monthDisplayNames) {
            System.out.printf("%12s", strName);
        }
        System.out.println();
        for (int i = 0; i < monthDisplayNames.length - 1; i++) {
            System.out.print("=======");
        }
        System.out.println("====");
        for (int i = 0; i < NUMBER_OF_CUSTOMERS; i++) {
            System.out.printf("Client ID:1%02d%2s", i, "|");
            for (int j = 0; j < NUMBER_OF_MONTHS; j++) {
                System.out.printf("%6d", salesMatrix[i][j]);
            }
            System.out.println();
        }
        System.out.println();
    }

    public static void main(String[] args) throws 
            InterruptedException, ExecutionException {
        generateMatrix();
        printMatrix();
        ExecutorService executor = Executors.newFixedThreadPool(10);
        Set<Future<Integer>> set = new HashSet<Future<Integer>>();
        for (int row = 0; row < NUMBER_OF_CUSTOMERS; row++) {
            Callable<Integer> callable = new Summer(row);
            Future<Integer> future = executor.submit(callable);
            set.add(future);
        }
        int sum = 0;
        for (Future<Integer> future : set) {
            sum += future.get();
        }
        System.out.printf("%nThe annual turnover (bags): %s%n%n", sum);
        executor.shutdown();
    }
}

部分结果:

                  一月          二月          三月          四月          五月          六月          七月          八月          九月          十月         十一月         十二月            

========================================================================================

Client ID:100 |    82    19    30    27    90    33    64    32    20    40    60    36

Client ID:101 |    19    99    14    26    87    86    26    22    51     2    75    57

Client ID:102 |    41    86    32    68    91    52     0    38    77    13    53     7

......

Client ID:197 |    68    93    72    72     8    68    10     6    90    11    81    78

Client ID:198 |    80    86    88    87    17    87    47    62    88    62    76    47

Client ID:199 |    10    38    74    36    75    17    31     4    48    92    43    59

Totaling for client 101 completed

Totaling for client 103 completed

Totaling for client 102 completed

......

Totaling for client 104 completed

Totaling for client 130 completed

Totaling for client 120 completed

The annual turnover (bags): 60275

FutureTask示例程序:演示股票交易程序,一个懒惰线程随时可以取消提交的任务,如果订单已经完成取消失败,如果任务正在执行且可以中断则取消该线程剩余的处理流程,如果订单已提交并且在分配给线程执行之前被取消,则订单会取消成功。

package test1;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class StocksOrderProcessor {
    public static final int MAX_NUMBER_OF_ORDERS = 1000;
    private static final ExecutorService executor = Executors
            .newFixedThreadPool(100);
    private static List<Future<Integer>> ordersToProcess = new ArrayList<Future<Integer>>();

    private static class OrderExecutor implements Callable<Integer> {
        private int id = 0;
        private int count = 0;

        public OrderExecutor(int id) {
            this.id = id;
        }

        @Override
        public Integer call() throws Exception {
            while (count < 50) {
                count++;
                Thread.sleep(new Random(System.currentTimeMillis() % 100)
                        .nextInt(10));
            }
            System.out.println("Successfully executed order: " + id);
            return id;
        }
    }

    private static void submitOrder(int id) {
        Callable<Integer> callable = new OrderExecutor(id);
        ordersToProcess.add(executor.submit(callable));
    }

    public static void main(String[] args) {
        System.out.printf("Submitting %d trades%n", MAX_NUMBER_OF_ORDERS);
        for (int i = 0; i < MAX_NUMBER_OF_ORDERS; i++) {
            submitOrder(i);
        }
        new Thread(new EvilThread(ordersToProcess)).start();
        System.out.println("Cancelling a few orders at random");
        try {
            // 为了让所有任务都可以完成,让执行器等待30秒钟
            executor.awaitTermination(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Checking status before shutdown");
        int count = 0;
        for (Future<Integer> future : ordersToProcess) {
            if (future.isCancelled()) {
                count++;
            }
        }
        System.out.printf("%d trades cancelled%n", count);
        // shutdown方法并不意味着之前提交的任务被立即取消,而是发起任务顺序停止以便之前提交的任务
        // 可以被执行,但是他保证不再接受新的任务。ExecuteExistingDelayedTaskAfterShutdownPolicy
        // 被设置为false意味着现有未完成的任务会被取消,反之,不会被取消。
        // ContinueExistingPeriodicTasksAfterShutdownPolicy
        // 被设置为true,那么已有的周期性任务会被取消。
        executor.shutdown();
    }
}

class EvilThread implements Runnable {
    private List<Future<Integer>> ordersToProcess;

    public EvilThread(List<Future<Integer>> futures) {
        this.ordersToProcess = futures;
    }

    @Override
    public void run() {
        Random myNextKill = new Random(System.currentTimeMillis() % 100);
        for (int i = 0; i < 100; i++) {
            int index = myNextKill
                    .nextInt(StocksOrderProcessor.MAX_NUMBER_OF_ORDERS);
            boolean cancel = ordersToProcess.get(index).cancel(true);
            if (cancel == true) {
                System.out.println("Cancel Order Succeed: " + index);
            } else {
                System.out.println("Cancel Order Failed: " + index);
            }
            try {
                Thread.sleep(myNextKill.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

部分执行结果:

Submitting 1000 trades

Successfully executed order: 64

Cancelling a few orders at random

Cancel Order Succeed: 857

Cancel Order Succeed: 402

Cancel Order Succeed: 262

Cancel Order Succeed: 72

Successfully executed order: 62

Successfully executed order: 102

Cancel Order Succeed: 327

......

Cancel Order Failed: 539

本文转载自:

共有 人打赏支持
小菜鸡1
粉丝 10
博文 59
码字总数 16851
作品 0
深圳
程序员
Executor框架结构与主要成员(一)

本文分两部分来介绍Executor:Executor的结构和Executor框架包含的成员组件 1、Executor框架的结构 Executor主要由3大部分组成。 1.1、任务。包含被执行任务需要实现的接口:Runnable接口或C...

Dreyer
2016/05/07
133
0
Java多线程、并发杂记

多线程涉及的类可以分为以下几类: 可执行对象:最基本的多线程 执行器:简化多线程编程 工具类 容器 并发控制 一、可执行对象: 1、Runnable: 执行单位:Thread 创建线程的两种方式(来自于...

rathan0
2016/02/17
60
0
java Callable & Future & FutureTask

实现Runnable接口的线程类与一个缺陷,就是在任务执行完之后无法取得任务的返回值。 如果需要获取执行结果,就必须通过共享变量或者使用线程通信的方式来达到效果,这样使用起来就比较麻烦 ...

Key_Stone
2016/09/16
20
0
Java并发编程原理与实战三十一:Future&FutureTask 浅析

一、Futrue模式有什么用? ------>正所谓技术来源与生活,这里举个栗子。在家里,我们都有煮菜的经验。(如果没有的话,你们还怎样来泡女朋友呢?你懂得)。现在女票要你煮四菜一汤,这汤是鸡...

pony1223
08/19
0
0
Thread,Runnable,Callable. 多线程

编写多线程程序是为了实现多任务的并发执行,从而能够更好地与用户交互。一般有三种方法,Thread,Runnable,Callable.   Runnable和Callable的区别是,   (1)Callable规定的方法是call(...

千惊万喜
2016/06/30
12
0

没有更多内容

加载失败,请刷新页面

加载更多

ehlib 修改 使行号字体颜色 与标题字体颜色 一致

对ehlib 显示效果不够满意,而做的调整 修改这个过程:procedure TCustomDBGridEh.DrawIndicatorCell(ACol, ARow: Longint; AreaCol, AreaRow: Longint; ARect: TRect; AState: TGri......

vga
36分钟前
0
0
Bash重定向详解

Bash重定向详解 Bash的重定向指的是将命令的输入和输出导向不同地方,而不是默认的标准输入、标准输出和标准错误。Bash的重定向实际上是对标准输入、标准输出和标准错误的重置,进而将所需输...

小陶小陶
今天
3
0
EventBus原理深度解析

一、问题描述 在工作中,经常会遇见使用异步的方式来发送事件,或者触发另外一个动作:经常用到的框架是MQ(分布式方式通知)。如果是同一个jvm里面通知的话,就可以使用EventBus。由于Event...

yangjianzhou
今天
23
0
OpenCV图像处理实例:libuv+cvui显示摄像头视频

#include <iostream>#include <opencv2/opencv.hpp>#define CVUI_IMPLEMENTATION#include <cvui.h>extern "C"{#include <uv.h>}using namespace std;#define WINDOW_NAM......

IOTService
今天
4
0
openJDK之JDK9的String

1.openJDK8的String 先来看下openJDK8的String的底层,如下图1.1所示: 图1.1 底层上使用的是char[],即char数组 每个char占16个bit,Character.SIZE的值是16。 2.openJDK9中的String 图2.1...

克虏伯
今天
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部