Java多线程-Future

发表于 | 分类于 ，

Future模式

  • Future模式是多线程开发中的常见的一种设计模式，核心思想是异步调用。当我们调用一个函数方法时，如果函数执行的很慢，需要等待。很多情况上，我们不需要立刻获取结果，而是让函数方法先返回，后台去计算结果，等到需要使用时，再去尝试获取数据。

Future接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public interface Future<V> {
    //试图取消对此任务的执行。如果任务已完成、或已取消，或者由于某些其他原因而无法取消，则此尝试将失败。当调用 cancel 时，如果调用成功，而此任务尚未启动，则此任务将永不运行。如果任务已经启动，则 mayInterruptIfRunning 参数确定是否应该以试图停止任务的方式来中断执行此任务的线程。此方法返回后，对 isDone() 的后续调用将始终返回 true。如果此方法返回 true，则对 isCancelled()的后续调用将始终返回 true。
    boolean cancel(boolean mayInterruptIfRunning);
    // 任务正常完成前取消，返回true
    boolean isCancelled();
    // 任务是否正常结束
    boolean isDone();
    // 等待线程结果的返回，阻塞线程
    V get() throws InterruptedException, ExecutionException;
    // 设置超时时间
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask

  • FutureTask类是Future 的一个实现，并实现了RunnableFuture
  • 可通过Excutor(线程池) 来执行,也可传递给Thread对象执行。
  • 如果在主线程中需要执行比较耗时的操作时，但又不想阻塞主线程时，可以把这些作业交给Future对象在后台完成，当主线程将来需要时，就可以通过Future对象获得后台作业的计算结果或者执行状态。
  • Executor框架利用FutureTask来完成异步任务，并可以用来进行任何潜在的耗时的计算。一般FutureTask多用于耗时的计算，主线程可以在完成自己的任务后，再去获取结果。
  • FutureTask类既可以使用new Thread(Runnable r)放到一个新线程中跑，也可以使用ExecutorService.submit(Runnable r)放到线程池中跑，而且两种方式都可以获取返回结果，但实质是一样的，即如果要有返回结果那么构造函数一定要注入一个Callable对象。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
public class FutureTask<V> implements RunnableFuture<V> {
    // 任务状态，及状态转换：
    // * NEW -> COMPLETING -> NORMAL
    // * NEW -> COMPLETING -> EXCEPTIONAL
    // * NEW -> CANCELLED
    // * NEW -> INTERRUPTING -> INTERRUPTED
    private volatile int state;
    private static final int NEW          = 0;
    private static final int COMPLETING   = 1;
    private static final int NORMAL       = 2;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED  = 6;
    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    private volatile WaitNode waiters;
    /**
     * 构造方法，传入Calable接口实现类
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    /**
    * 构造方法，传入Runnable接口实现类
    */
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }
    // 。。。。
}

应用实例

Future模式的参与者：

参与者 作用
Main 系统启动，调用Client发出请求
Client 返回Data对象，立即返回FutureData，并开启ClientThread线程装配RealData
FutureData 返回数据的接口
RealData 真实数据

Date接口

1
2
3
public interface Data {
    public String getResult();
}

FutureData

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class FutureData implements Data{
    protected RealData realData = null;
    protected boolean isReady = false;
    public synchronized void setRealData(RealData realData) {
        if (isReady) {
            return;
        }
        this.realData = realData;
        isReady = true;
        notifyAll();
    }
    @Override
    public String getResult() {
        while (!isReady) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return realData.result;
    }
}

RealData

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class RealData implements Data{
    protected final String result;
    public RealData(String result) {
        StringBuffer sb = new StringBuffer();
        for (int i = 0; i < 10; i++) {
            sb.append(result);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.result = sb.toString();
    }
    @Override
    public String getResult() {
        return result;
    }
}

client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public class Main {
    public static void main(String[] args) {
        Client client = new Client();
        //会立即返回，得到FutureData
        Data data = client.request("name");
        System.out.println("请求完毕");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("数据 = " + data.getResult());
    }
}
public class Client {
    public Data request(final String queryStr) {
        final FutureData futureData = new FutureData();
        new Thread(() -> {
            RealData realData = new RealData(queryStr);
            futureData.setRealData(realData);
        }).start();
        return futureData;
    }
}
Donate comment here