java并发Future、FutureTask、Callable
2019-07-06 本文已影响0人
small瓜瓜
在发生
IO事件的时候，CPU总是要等待数据进入内存，这是非常低效的，主线程在这段时间内堵塞无法进行其他的操作。为了提高运行效率，在java中我们可以创建多线程来避免主线程的等待。
Java创建线程执行任务一般的方式有2种：
- 通过继承
Thread类并重写其run方法 - 通过实现
Runnable接口并重写其run方法
当时这两种方式都不法获取运行的结果，这时候很多场景都不太适用了
有一个比较长
int类型的数组，现在要计算它的所有元素的总和。为了提高效率，要求使用多线程进行计算。
如果有一个这样的需求，就必须要多线程执行并获取他们的运算结果了。为了能完成上面的要求，这里就需要对传统的方式进行改进。具体改进如下：
class CalcSubSum implements Runnable {
private int offset;
private int count;
private volatile int result;
private int[] as;
private volatile boolean isSuccess;
public CalcSubSum(int offset, int count, int[] as) {
this.offset = offset;
this.count = count;
this.as = as;
}
@Override
public void run() {
int end = offset + count;
int sum = 0;
for (int i = offset; i < end; i++) {
sum += as[i];
}
isSuccess = true;
result = sum;
}
public int getResult() {
while (!isSuccess) ;
return result;
}
}
上面通过实现Runnable接口创建了一个线程，通过该类中的result成员属性获取运算的结果。为了避免可见性问题，result和isSuccess都加上了volatile关键字修饰。
那就看看是不是可以实现吧，编写主方法如下：
public class Main {
private static int threadNum = 3;
public static void main(String[] args) {
int[] bigArray = {
1, 2, 3, 4, 5, 6, 7, 8, 9, 10
};
int subLen = bigArray.length / threadNum;
int endSubLen = subLen + bigArray.length % threadNum;
List<CalcSubSum> calcSubSumList = new ArrayList<>();
for (int i = 0; i < threadNum; i++) {
int offset = i * subLen;
int len = subLen;
if (i == threadNum - 1) {
len = endSubLen;
}
CalcSubSum calcSubSum = new CalcSubSum(offset, len, bigArray);
new Thread(calcSubSum).start();
calcSubSumList.add(calcSubSum);
}
int sum = 0;
for (CalcSubSum calcSubSum : calcSubSumList) {
int result = calcSubSum.getResult();
System.out.println("result = " + result);
sum += result;
}
System.out.println("sum = " + sum);
}
}
运行上面的代码可以正确获取结果：
result = 6
result = 15
result = 34
sum = 55
程序的执行效率是提高了，但是如果需求有变那个类就需要改写了，比如计算所有元素的乘积，且要求返回值为long类型。这样就不太容易扩展，不符合开闭原则。同时在获取线程运算的结果时，使用了循环判断的方式这是非常消耗CPU的，这种等待方式一点都不优雅。有了问题那现在就是解决问题了，使用组合的方式将执行的逻辑提取出来。
下面的注释我用-代替，因为用//注释文字有点不太清晰
- 真正要执行任务的接口，泛型T是要返回值得类型，基本数据类型只能使用其包装类
interface Task<T> {
T doTask();
}
- 用获取执行结果的操作类
class CalcSub<T> implements Runnable {
private volatile T result;
private volatile boolean isSuccess;
private Task<T> task;
public CalcSub(Task<T> task) {
this.task = task;
}
@Override
public void run() {
result = run0();
isSuccess = true;
}
T run0() {
if (task != null)
return task.doTask();
throw new RuntimeException("线程中没有要执行的任务！！");
}
public T getResult() {
while (!isSuccess) ;
return result;
}
}
- Task接口的一个实现，一个具体可以获取子数组和的任务类
class CalcSubSum implements Task<Integer> {
private int offset;
private int count;
private int[] as;
public CalcSubSum(int offset, int count, int[] as) {
this.offset = offset;
this.count = count;
this.as = as;
}
@Override
public Integer doTask() {
int end = offset + count;
int sum = 0;
for (int i = offset; i < end; i++) {
sum += as[i];
}
return sum;
}
}
这样就可以很方便的扩展出新的任务而不用修改原代码了。
现在我们添加一个计算乘积结果的任务类
class CalcSubMultiply implements Task<Long> {
private int offset;
private int count;
private int[] as;
public CalcSubMultiply(int offset, int count, int[] as) {
this.offset = offset;
this.count = count;
this.as = as;
}
@Override
public Long doTask() {
int end = offset + count;
long mul = 0;
for (int i = offset; i < end; i++) {
mul *= as[i];
}
return mul;
}
}
但这里可扩展已经全部搞定了，我们就继续解决剩下的问题吧。getResult()不优雅等问题，这里我们可以采用线程等待的方式，实现如下：
class CalcSub<T> implements Runnable {
private volatile T result;
private volatile boolean isSuccess;
private Task<T> task;
private List<Thread> threads;
public CalcSub(Task<T> task) {
this.task = task;
threads = new Vector<>();
}
@Override
public void run() {
result = run0();
isSuccess = true;
done();
}
- 这个方法可以作为完成回调重写，但是一定要调用父类的super.done()
public void done() {
for (Thread thread : threads) {
LockSupport.unpark(thread);
}
}
T run0() {
if (task != null)
return task.doTask();
throw new RuntimeException("");
}
- 将调用该方法的线程进行堵塞处理
public T getResult() {
Thread thread = Thread.currentThread();
if (!isSuccess && !threads.contains(thread)) {
threads.add(thread);
LockSupport.park(thread);
}
return result;
}
}
这个类的功能还是比较简陋的，只要一个isSuccess判断是否成功，状态变化单调。接下来我们就为他加入一些状态：
1. 创建时状态：NEW
2. 运行时状态：COMPLETING
3. 成功完成后状态：NORMAL
4. 发生异常后状态：EXCEPTIONAL
5. 在未启动时取消的状态：CANCELLED
7. 中断时状态：INTERRUPTING
8. 中断时状态：INTERRUPTED
然后根据相应的状态写好逻辑就可以了。CalcSub写好后，又有一个问题，如果需要写很多个这样的线程任务控制类，这些状态转移的方法名称和其他获取结果方法的名称等就比较伤脑筋了。我们怎么去命名我们的方法名称呢？使我们写的框架或是代码其他人一看就可以明白呢？这里我们就要定一个规范，通过接口实现这样的规范。那个方法是什么功能写清楚，具体实现让使用者去完成。
具体修改如下：
- 这是一套规范接口
interface FutureNorm<T> {
// 取消
boolean cancel(boolean mayInterruptIfRunning);
// 判断是否取消
boolean isCancelled();
// 判断是否完成
boolean isDone();
// 获取结果，如果没有完成就堵塞，直到运算结束
T get() throws InterruptedException, ExecutionException;
}
class CalcSub<T> implements Runnable, FutureNorm<T> {
private volatile T result;
private volatile boolean isSuccess;
private Task<T> task;
private List<Thread> threads;
public CalcSub(Task<T> task) {
this.task = task;
threads = new Vector<>();
}
@Override
public void run() {
result = run0();
isSuccess = true;
done();
}
public void done() {
for (Thread thread : threads) {
LockSupport.unpark(thread);
}
}
T run0() {
if (task != null)
return task.doTask();
throw new RuntimeException("");
}
public T getResult() {
Thread thread = Thread.currentThread();
if (!isSuccess && !threads.contains(thread)) {
threads.add(thread);
LockSupport.park(thread);
}
return result;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return isSuccess;
}
@Override
public boolean isCancelled() {
return isSuccess;
}
@Override
public boolean isDone() {
return isSuccess;
}
@Override
public T get() throws InterruptedException, ExecutionException {
return getResult();
}
}
本文说要简析浅谈Future、FutureTask、Callable。但是直到现在都没有说相关的内容，或许让您有些疑惑。前面的铺垫已经好了，按我们现在开始讲解吧！
直接上Future的源码：
/*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
package java.util.concurrent;
/**
* A {@code Future} represents the result of an asynchronous
* computation. Methods are provided to check if the computation is
* complete, to wait for its completion, and to retrieve the result of
* the computation. The result can only be retrieved using method
* {@code get} when the computation has completed, blocking if
* necessary until it is ready. Cancellation is performed by the
* {@code cancel} method. Additional methods are provided to
* determine if the task completed normally or was cancelled. Once a
* computation has completed, the computation cannot be cancelled.
* If you would like to use a {@code Future} for the sake
* of cancellability but not provide a usable result, you can
* declare types of the form {@code Future<?>} and
* return {@code null} as a result of the underlying task.
*
* <p>
* <b>Sample Usage</b> (Note that the following classes are all
* made-up.)
* <pre> {@code
* interface ArchiveSearcher { String search(String target); }
* class App {
* ExecutorService executor = ...
* ArchiveSearcher searcher = ...
* void showSearch(final String target)
* throws InterruptedException {
* Future<String> future
* = executor.submit(new Callable<String>() {
* public String call() {
* return searcher.search(target);
* }});
* displayOtherThings(); // do other things while searching
* try {
* displayText(future.get()); // use future
* } catch (ExecutionException ex) { cleanup(); return; }
* }
* }}</pre>
*
* The {@link FutureTask} class is an implementation of {@code Future} that
* implements {@code Runnable}, and so may be executed by an {@code Executor}.
* For example, the above construction with {@code submit} could be replaced by:
* <pre> {@code
* FutureTask<String> future =
* new FutureTask<String>(new Callable<String>() {
* public String call() {
* return searcher.search(target);
* }});
* executor.execute(future);}</pre>
*
* <p>Memory consistency effects: Actions taken by the asynchronous computation
* <a href="package-summary.html#MemoryVisibility"> <i>happen-before</i></a>
* actions following the corresponding {@code Future.get()} in another thread.
*
* @see FutureTask
* @see Executor
* @since 1.5
* @author Doug Lea
* @param <V> The result type returned by this Future's {@code get} method
*/
public interface Future<V> {
/**
* Attempts to cancel execution of this task. This attempt will
* fail if the task has already completed, has already been cancelled,
* or could not be cancelled for some other reason. If successful,
* and this task has not started when {@code cancel} is called,
* this task should never run. If the task has already started,
* then the {@code mayInterruptIfRunning} parameter determines
* whether the thread executing this task should be interrupted in
* an attempt to stop the task.
*
* <p>After this method returns, subsequent calls to {@link #isDone} will
* always return {@code true}. Subsequent calls to {@link #isCancelled}
* will always return {@code true} if this method returned {@code true}.
*
* @param mayInterruptIfRunning {@code true} if the thread executing this
* task should be interrupted; otherwise, in-progress tasks are allowed
* to complete
* @return {@code false} if the task could not be cancelled,
* typically because it has already completed normally;
* {@code true} otherwise
*/
boolean cancel(boolean mayInterruptIfRunning);
/**
* Returns {@code true} if this task was cancelled before it completed
* normally.
*
* @return {@code true} if this task was cancelled before it completed
*/
boolean isCancelled();
/**
* Returns {@code true} if this task completed.
*
* Completion may be due to normal termination, an exception, or
* cancellation -- in all of these cases, this method will return
* {@code true}.
*
* @return {@code true} if this task completed
*/
boolean isDone();
/**
* Waits if necessary for the computation to complete, and then
* retrieves its result.
*
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
*/
V get() throws InterruptedException, ExecutionException;
/**
* Waits if necessary for at most the given time for the computation
* to complete, and then retrieves its result, if available.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
* @throws TimeoutException if the wait timed out
*/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
Future是java并发包juc下的一个接口，它定义了如何从异步操作中获取线程状态或运行结果的一套规范。
他的作用就是对应前面的我们所写的FutureNorm接口。
在看看Callable的源码：
/**
* A task that returns a result and may throw an exception.
* Implementors define a single method with no arguments called
* {@code call}.
*
* <p>The {@code Callable} interface is similar to {@link
* java.lang.Runnable}, in that both are designed for classes whose
* instances are potentially executed by another thread. A
* {@code Runnable}, however, does not return a result and cannot
* throw a checked exception.
*
* <p>The {@link Executors} class contains utility methods to
* convert from other common forms to {@code Callable} classes.
*
* @see Executor
* @since 1.5
* @author Doug Lea
* @param <V> the result type of method {@code call}
*/
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
这又是一个接口，只有一个call方法，通过注释就可以知道它和我们写的Task作用是一样的
接着再看看FutureTask的源码：
package java.util.concurrent;
import java.util.concurrent.locks.LockSupport;
/**
* A cancellable asynchronous computation. This class provides a base
* implementation of {@link Future}, with methods to start and cancel
* a computation, query to see if the computation is complete, and
* retrieve the result of the computation. The result can only be
* retrieved when the computation has completed; the {@code get}
* methods will block if the computation has not yet completed. Once
* the computation has completed, the computation cannot be restarted
* or cancelled (unless the computation is invoked using
* {@link #runAndReset}).
*
* <p>A {@code FutureTask} can be used to wrap a {@link Callable} or
* {@link Runnable} object. Because {@code FutureTask} implements
* {@code Runnable}, a {@code FutureTask} can be submitted to an
* {@link Executor} for execution.
*
* <p>In addition to serving as a standalone class, this class provides
* {@code protected} functionality that may be useful when creating
* customized task classes.
*
* @since 1.5
* @author Doug Lea
* @param <V> The result type returned by this FutureTask's {@code get} methods
*/
public class FutureTask<V> implements RunnableFuture<V> {
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races. Sync control in the current design relies
* on a "state" field updated via CAS to track completion, along
* with a simple Treiber stack to hold waiting threads.
*
* Style note: As usual, we bypass overhead of using
* AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
*/
/**
* The run state of this task, initially NEW. The run state
* transitions to a terminal state only in methods set,
* setException, and cancel. During completion, state may take on
* transient values of COMPLETING (while outcome is being set) or
* INTERRUPTING (only while interrupting the runner to satisfy a
* cancel(true)). Transitions from these intermediate to final
* states use cheaper ordered/lazy writes because values are unique
* and cannot be further modified.
*
* Possible state transitions:
* 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;
/**
* Returns result or throws exception for completed task.
*
* @param s completed state value
*/
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Callable}.
*
* @param callable the callable task
* @throws NullPointerException if the callable is null
*/
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Runnable}, and arrange that {@code get} will return the
* given result on successful completion.
*
* @param runnable the runnable task
* @param result the result to return on successful completion. If
* you don't need a particular result, consider using
* constructions of the form:
* {@code Future<?> f = new FutureTask<Void>(runnable, null)}
* @throws NullPointerException if the runnable is null
*/
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
/**
* @throws CancellationException {@inheritDoc}
*/
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
/**
* @throws CancellationException {@inheritDoc}
*/
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}
/**
* Protected method invoked when this task transitions to state
* {@code isDone} (whether normally or via cancellation). The
* default implementation does nothing. Subclasses may override
* this method to invoke completion callbacks or perform
* bookkeeping. Note that you can query status inside the
* implementation of this method to determine whether this task
* has been cancelled.
*/
protected void done() { }
/**
* Sets the result of this future to the given value unless
* this future has already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon successful completion of the computation.
*
* @param v the value
*/
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
/**
* Causes this future to report an {@link ExecutionException}
* with the given throwable as its cause, unless this future has
* already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon failure of the computation.
*
* @param t the cause of failure
*/
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
/**
* Ensures that any interrupt from a possible cancel(true) is only
* delivered to a task while in run or runAndReset.
*/
private void handlePossibleCancellationInterrupt(int s) {
// It is possible for our interrupter to stall before getting a
// chance to interrupt us. Let's spin-wait patiently.
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
// assert state == INTERRUPTED;
// We want to clear any interrupt we may have received from
// cancel(true). However, it is permissible to use interrupts
// as an independent mechanism for a task to communicate with
// its caller, and there is no way to clear only the
// cancellation interrupt.
//
// Thread.interrupted();
}
/**
* Simple linked list nodes to record waiting threads in a Treiber
* stack. See other classes such as Phaser and SynchronousQueue
* for more detailed explanation.
*/
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
/**
* Awaits completion or aborts on interrupt or timeout.
*
* @param timed true if use timed waits
* @param nanos time to wait, if timed
* @return state upon completion
*/
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
}
为了让看起来更方便，FutureTask源码部分方法进行了删除。其实这个就是我们写的CalcSub类。
FutureTask和CalcSub的继承图如下：
上面写的代码主要是提供一个思路，并通过实现其部分功能来进行说明其作用，如果想真正了解Future、FutureTask、Callable还是需要看源码的。
代码主要是提供思路，并没有对安全问题进行处理
