Hello everyone, I am Glacier ~~
This article is a bit long, but full of dry goods, in the form of a practical case analysis of two asynchronous models, and from the source perspective of the depth of Future interface and FutureTask class, I hope you step down to the heart, open your IDE, follow the article to see the source code, I believe you must gain a lot!
One or two asynchronous models
In Java concurrent programming, there are roughly two asynchronous programming models. One is to run other tasks in parallel in an asynchronous manner without returning the result data of the task. One runs other tasks asynchronously and needs to return results.
1. Asynchronous model with no results returned
If an asynchronous task does not return a result, the task can be directly thrown into a thread or thread pool. In this case, the execution result data of the task cannot be obtained directly. One way is to use the callback method to obtain the running result of the task.
The specific scheme is: define a callback interface, and in the interface to define the method to receive the result of the task data, the specific logic in the callback interface implementation class. The callback interface is run in a thread or thread pool along with the task parameters. After the task runs, the interface methods are called and the logic in the callback interface implementation class is executed to process the resulting data. Here, give a simple example for reference.
- Define the callback interface
package io.binghe.concurrent.lab04;
/ * * *@author binghe
* @version 1.0.0
* @descriptionDefine the callback interface */
public interface TaskCallable<T> {
T callable(T t);
}
Copy the codeGenerics for interfaces, where generics are defined for callback interfaces.
- Defines a wrapper class for task result data
package io.binghe.concurrent.lab04;
import java.io.Serializable;
/ * * *@author binghe
* @version 1.0.0
* @descriptionResult */
public class TaskResult implements Serializable {
private static final long serialVersionUID = 8678277072402730062L;
/** * Task status */
private Integer taskStatus;
/** * Task message */
private String taskMessage;
/**
* 任务结果数据
*/
private String taskResult;
// omit the getter and setter methods
@Override
public String toString(a) {
return "TaskResult{" +
"taskStatus=" + taskStatus +
", taskMessage='" + taskMessage + '\' ' +
", taskResult='" + taskResult + '\' ' +
'} '; }}Copy the code- Create an implementation class for the callback interface
The implementation class of the callback interface is mainly used for business processing of the result returned by the task. Here, for demonstration purposes, only the result data is returned. You need to analyze and handle the problem according to specific business scenarios.
package io.binghe.concurrent.lab04;
/ * * *@author binghe
* @version 1.0.0
* @descriptionThe implementation class for the callback function */
public class TaskHandler implements TaskCallable<TaskResult> {
@Override
public TaskResult callable(TaskResult taskResult) {
//TODO takes the result data and processes it further
System.out.println(taskResult.toString());
returntaskResult; }}Copy the code- Create an execution class for the task
The execution class of a task is a class that concretely performs the task, implements the Runnable interface, defines a member variable of type callback interface and a task parameter of type String (simulated task parameters) in this class, and injects the callback interface and task parameters into the constructor. The task is executed in the RUN method, and when the task is complete, the result data of the task is encapsulated as a TaskResult object. The TaskResult object is passed to the callback method by calling the callback interface method.
package io.binghe.concurrent.lab04;
/ * * *@author binghe
* @version 1.0.0
* @descriptionTask execution class */
public class TaskExecutor implements Runnable{
private TaskCallable<TaskResult> taskCallable;
private String taskParameter;
public TaskExecutor(TaskCallable<TaskResult> taskCallable, String taskParameter){
this.taskCallable = taskCallable;
this.taskParameter = taskParameter;
}
@Override
public void run(a) {
//TODO a series of business logic that encapsulates the result data into TaskResult objects and returns them
TaskResult result = new TaskResult();
result.setTaskStatus(1);
result.setTaskMessage(this.taskParameter);
result.setTaskResult("Asynchronous callback successful"); taskCallable.callable(result); }}Copy the codeNow that the big framework is complete, it’s time to test to see if you can get the results of an asynchronous task.
- Asynchronous task test class
package io.binghe.concurrent.lab04;
/ * * *@author binghe
* @version 1.0.0
* @descriptionTest the callback */
public class TaskCallableTest {
public static void main(String[] args){
TaskCallable<TaskResult> taskCallable = new TaskHandler();
TaskExecutor taskExecutor = new TaskExecutor(taskCallable, "Test callback task");
newThread(taskExecutor).start(); }}Copy the codeIn the test class, use the Thread class to create a new Thread and start the Thread to run the task. The final interface data for running the program is shown below.
TaskResult{taskStatus=1, taskMessage='Test callback task', taskResult='Asynchronous callback successful'}
Copy the codeYou can take a look at this way of getting asynchronous results. Here, we simply use the Thread class to create and start a Thread, or we can use a Thread pool. You can implement your own way of getting asynchronous results through the callback interface in a thread pool.
2. Asynchronous model with results returned
Although it is possible to get the results of an asynchronous task using the callback interface, it is a bit more complicated to use. A processing scheme is provided in the JDK that can directly return asynchronous results. The most common is to use the Future interface or its implementation class, FutureTask, to receive the results of a task.
- Use the Future interface to get asynchronous results
The Future interface is often used in conjunction with thread pools to obtain asynchronous execution results, as shown below.
package io.binghe.concurrent.lab04;
import java.util.concurrent.*;
/ * * *@author binghe
* @version 1.0.0
* @descriptionTest Future to get asynchronous results */
public class FutureTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<String> future = executorService.submit(new Callable<String>() {
@Override
public String call(a) throws Exception {
return "Test Future for asynchronous results"; }}); System.out.println(future.get()); executorService.shutdown(); }}Copy the codeThe result is as follows.
Test the Future to get asynchronous resultsCopy the code- Use the FutureTask class to get asynchronous results
The FutureTask class can be used in conjunction with either a Thread class or a Thread pool. Next, let’s look at the two uses.
An example of use with the Thread class is shown below.
package io.binghe.concurrent.lab04;
import java.util.concurrent.*;
/ * * *@author binghe
* @version 1.0.0
* @descriptionTest FutureTask to get asynchronous results */
public class FutureTaskTest {
public static void main(String[] args)throws ExecutionException, InterruptedException{
FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
@Override
public String call(a) throws Exception {
return "Test FutureTask for asynchronous results"; }});newThread(futureTask).start(); System.out.println(futureTask.get()); }}Copy the codeThe result is as follows.
Test FutureTask for asynchronous resultsCopy the codeAn example of use with thread pools is shown below.
package io.binghe.concurrent.lab04;
import java.util.concurrent.*;
/ * * *@author binghe
* @version 1.0.0
* @descriptionTest FutureTask to get asynchronous results */
public class FutureTaskTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newSingleThreadExecutor();
FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
@Override
public String call(a) throws Exception {
return "Test FutureTask for asynchronous results"; }}); executorService.execute(futureTask); System.out.println(futureTask.get()); executorService.shutdown(); }}Copy the codeThe result is as follows.
Test FutureTask for asynchronous resultsCopy the codeYou can see how much easier it is to get asynchronous results using the Future interface or FutureTask class than using the callback interface. Note: There are many ways to implement asynchrony, but this is just using multithreading as an example.
Next, take a closer look at the Future interface.
Second, in-depth analysis of the Future interface
1. The Future interface
Future is a new asynchronous programming interface in JDK1.5, and its source code is shown below.
package java.util.concurrent;
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled(a);
boolean isDone(a);
V get(a) throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
Copy the codeAs you can see, a total of five abstract methods are defined in the Future interface. Here are the implications of each of these five methods.
- cancel(boolean)
Cancel the execution of the task, receiving a Boolean argument, returning true if the task is canceled successfully, false otherwise. If the task is completed, completed, or cannot be canceled for some other reason, the method returns false, indicating that the task cancellation failed. This method is called when the task is not started and returns true (cancelled successfully), the current task is no longer running. If the task is already started, the currently running task is canceled by interrupting the currently running thread based on the currently passed Boolean argument.
- isCancelled()
Determines whether the task was canceled before completion, and returns true if the task was canceled before completion; Otherwise, return false.
Note one detail here: If the task is not started or cancelled before completion, it will return true indicating that the task has been cancelled successfully. All other cases return false.
- isDone()
Check whether the task is complete. If the task ends, an exception is thrown, or is cancelled, the system returns true, indicating that the task is complete.
- get()
When the task is completed, the result data of the task is directly returned. When the task is not complete, wait for the task to complete and return the result data of the task.
- get(long, TimeUnit)
When the task is completed, the result data of the task is directly returned. When the task is not completed, the system waits for the task to complete and sets the timeout waiting time. If the task is completed within the timeout period, the result is returned. Otherwise, a TimeoutException is thrown.
2. RunnableFuture interface
The Future interface has an important subinterface, which is the RunnableFuture interface. The RunnableFuture interface inherits not only the Future interface, but also the java.lang.Runnable interface, whose source code is shown below.
package java.util.concurrent;
public interface RunnableFuture<V> extends Runnable.Future<V> {
void run(a);
}
Copy the codeHere, how many abstract methods are there in the RunnableFuture interface? Think it over! Hahaha…
This interface is relatively simple. The run() method is the method that is called when the task is run.
3. FutureTask class
The FutureTask class is a very important implementation of the RunnableFuture interface. It implements all the methods of the RunnableFuture interface, Future interface, and Runnable interface. The FutureTask class has a lot of source code, so I don’t want to paste it, but I want to check it out under java.util.concurrent.
(1) Variables and constants in FutureTask
The FutureTask class first defines a state variable, state, which is modified with the volatile keyword. All you need to know is that volatile implements thread safety through memory barriers and disallows reordering optimizations. A separate in-depth look at how the volatile keyword keeps threads safe will follow. Next, several task run time state constants are defined, as shown below.
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;
Copy the codeSeveral possible state change flows are presented in the code comments, as shown below.
NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED
Copy the codeNext, several other member variables are defined, as shown below.
private Callable<V> callable;
private Object outcome;
private volatile Thread runner;
private volatile WaitNode waiters;
Copy the codeThe familiar Callable interface is used to call the Call () method to perform specific tasks.
- Outcome: Object, indicating the result data or exception information obtained through get().
- Runner: a thread that runs a Callable and uses CAS to ensure thread safety. We only need to know that CAS is a way for Java to ensure thread safety. We will analyze in depth how CAS ensures thread safety in subsequent articles.
- Waiters: A variable of type WaitNode that represents a stack of waiting threads, which in FutureTask’s implementation exchanges the running state of the task through a CAS.
Take a look at the definition of the WaitNode class, shown below.
static final class WaitNode {
volatile Thread thread;
volatileWaitNode next; WaitNode() { thread = Thread.currentThread(); }}Copy the codeAs you can see, the WaitNode class is the static inner class of the FutureTask class, which defines a Thread member variable and a reference to the next WaitNode node. The thread variable is set to the current thread through the constructor.
(2) Construction method
Next, there are two simpler constructors for FutureTask, as shown below.
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW;
}
Copy the code(3) Whether to cancel and complete the method
Looking down at the source code, see methods for whether a task is cancelled, and methods for whether a task is completed, as shown below.
public boolean isCancelled(a) {
return state >= CANCELLED;
}
public boolean isDone(a) {
returnstate ! = NEW; }Copy the codeIn both of these methods, the task is judged by judging the state of the task to determine whether the task has been canceled and completed. Why is that? A second look at the state constants defined in the FutureTask class shows that they are defined regularly and not arbitrarily. Constants greater than or equal to CANCELLED are CANCELLED, INTERRUPTED, and INTERRUPTED. These three states can indicate that the thread has been CANCELLED. When the status is not equal to NEW, the task is complete.
Here’s what you can learn: Later in the process of coding, according to law to define the state of their own use, especially relates to have frequent state changes in the business operation, regular state can make business processing get twice the result with half the effort, it is also can learn by watching someone else’s source design, here, suggest you still see other people write excellent open source framework source code.
(4) Cancellation method
Moving down the source code, we see the cancel(Boolean) method, as shown below.
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;
}
Copy the codeNext, dismantle the cancel(Boolean) method. In the Cancel (Boolean) method, the status of the task and the result of the CAS operation are determined first. If the status of the task is not equal to NEW or the CAS operation returns false, false is returned, indicating that the task has failed to cancel. As shown below.
if(! (state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
Copy the codeNext, in the try block, you first determine whether you can abort the task by interrupting the thread in which it is currently running. If it is possible to interrupt the Thread of the current task, a Thread variable is used to point to the Thread running the task. If the variable is not null, the Thread is interrupted by calling interrupt() on the Thread object. Finally, the Thread is marked as interrupted. As shown below.
try {
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if(t ! =null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); }}}Copy the codeUnsafe.putorderedint () is the unsafe.putorderedInt () method used to change task state. Click in and see, as shown below.
public native void putOrderedInt(Object var1, long var2, int var4);
Copy the codeAs you can see, this is another native method, but I’ll leave it out here and explain what these methods do in a future article.
Next, the Cancel (Boolean) method enters the finally code block, as shown below.
finally {
finishCompletion();
}
Copy the codeYou can see that the finishCompletion() method is called within the Finallly code block, which, as its name suggests, terminates the task, so let’s see how it is implemented. Click through to the finishCompletion() method, as shown below.
private void finishCompletion(a) {
// 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
}
Copy the codeIn the finishCompletion() method, you first define a for loop with a waiters null termination factor, in which the CAS operation is verified as a success, and the logic in the if condition is executed successfully. First, define a for spin loop, in which the thread in the WaitNode stack is woken up to complete its run. When the thread in the WaitNode stack finishes running, exit the outer for loop with break. Next, call the done() method. What the hell is the done() method? Click in and see, as shown below.
protected void done(a) {}Copy the codeAs you can see, the done() method is an empty body of methods, leaving it to subclasses to implement the specific business logic.
When our specific business needs to execute some additional business logic when canceling a task, we can override the implementation of the done() method in a subclass.
(5) Get () method
Moving on to the code of the FutureTask class, the FutureTask class implements two GET () methods, as shown below.
public V get(a) throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false.0L);
return report(s);
}
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);
}
Copy the codeThe get() method, which has no arguments, blocks until the result of the task is returned while the task is running. The parameterized get() method raises a TimeoutException when the task is not completed and the wait time exceeds the timeout.
The main logic of the two get() methods is similar, one with no timeout setting and the other with timeout setting. Here’s the main logic. Call the awaitDone() method to determine whether the current state of the task is shorter or equal to COMPLETING those tasks, that is, the NEW state or COMPLETING those tasks, as shown below.
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); }}Copy the codeNext, unpack the awaitDone() method. In the awaitDone() method, the most important is the for spin loop, which first determines whether the current thread is interrupted, and if it is, calls removeWaiter() to remove the current thread from the stack and throws InterruptedException, as shown below.
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
Copy the codeNext, determine whether the current state of the task is complete. If so, and the stack handle is not empty, set the current thread in the stack to empty and return the current state of the task, as shown below.
int s = state;
if (s > COMPLETING) {
if(q ! =null)
q.thread = null;
return s;
}
Copy the codeWhen the task state is COMPLETING, cause the current thread to relinquish CPU resources, as shown below.
else if (s == COMPLETING)
Thread.yield();
Copy the codeIf the stack is empty, a stack object is created, as shown below.
else if (q == null)
q = new WaitNode();
Copy the codeIf the queued variable is false, assign queued through the CAS operation, and if the timed argument passed by the awaitDone() method is true, the timeout is calculated, and when the time has timed out, the current thread is removed from the stack and the task status is returned, as shown below. If not, reset the timeout, as shown below.
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);
}
Copy the codeIf all of the above conditions are not met, the current thread is set to the wait state, as shown below.
else
LockSupport.park(this);
Copy the codeNext, back in the get() method, when the awaitDone() method returns a result, or the status of the task does not meet the criteria, the report() method is called and the status of the current task is passed to the report() method and the result is returned, as shown below.
return report(s);
Copy the codeClick on the report() method to see how it is implemented, as shown below.
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);
}
Copy the codeAs can be seen, the implementation of the report() method is relatively simple. First, the outcome data is assigned to the x variable. Next, it mainly determines the status of the received task. CancellationException is raised when the status of the task is greater than or equal to CANCELLED, or ExecutionException is raised in other cases.
At this point, the analysis of the get() method is complete. Note: It’s important to understand the implementation of the Get () method, because the GET () method is one of the more frequent methods we use when working with the Future interface and FutureTask class.
(6) Set () method and setException() method
Moving on to the code of the FutureTask class, you see the set() and setException() methods, as shown below.
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion(); }}protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion(); }}Copy the codeSet () and setException() are set to NORMAL and exception () except that each set state is set to EXCEPTIONAL.
As for the finishCompletion() method, you’ve already analyzed it.
(7) Run () method and runAndReset() method
Next, there’s the run() method, whose source code is shown below.
public void run(a) {
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); }}Copy the codeIt can be said that whenever Future and FutureTask are used, the run() method will be called to run the task, and it is necessary to master the flow of the run() method. In the run() method, if the current state is not NEW, or the CAS operation returns false, it returns directly and no further logic is performed, as shown below.
if(state ! = NEW || ! UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
return;
Copy the codeNext, in the try code block, assign the member variable callable to a temporary variable C, determine that the temporary variable is not null, and the task status is NEW, call the Call () method of the Callable interface and receive the resulting data. And set the RAN variable to true. When the program throws an exception, the variable that receives the result is set to null, the RAN variable is set to false, and the setException() method is called to set the state of the task to EXCEPTIONA. Next, if the RAN variable is true, the set() method is called, as shown below.
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); }}Copy the codeNext, the program enters the finally code block, as shown below.
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);
}
Copy the codeHere, the runner is set to null, and the thread is interrupted if the current status of the task is greater than or equal to INTERRUPTING. Call the handlePossibleCancellationInterrupt () method, then, look at the handlePossibleCancellationInterrupt () method.
private void handlePossibleCancellationInterrupt(int s) {
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield();
}
Copy the codeAs you can see, handlePossibleCancellationInterrupt () method of implementation is simpler, when the state of the task for INTERRUPTING, use while () loop, the condition for the current task status for INTERRUPTING, To release CPU resources occupied by the current thread, that is, to release resources occupied by the thread when the task is completed.
The runAndReset() method has much the same logic as run(), except that the runAndReset() method resets the task state to NEW in the finally code block. The source code for the runAndReset() method is shown below and will not be repeated.
protected boolean runAndReset(a) {
if(state ! = NEW || ! UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if(c ! =null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch(Throwable ex) { setException(ex); }}}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
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
Copy the code(8) removeWaiter()
The removeWaiter() method, which mainly uses a spin loop to remove WaitNode threads, is relatively simple, as shown below.
private void removeWaiter(WaitNode node) {
if(node ! =null) {
node.thread = null;
retry:
for (;;) { // restart on removeWaiter race
for (WaitNode pred = null, q = waiters, s; q ! =null; q = s) {
s = q.next;
if(q.thread ! =null)
pred = q;
else if(pred ! =null) {
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if(! UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break; }}}Copy the codeFinally, at the end of the FutureTask class, there is the following code.
// 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 newError(e); }}Copy the codeThe purpose of this code will be explained in more detail in a future in-depth parsing CAS article, which is not discussed here.
At this point, the source code for the Future interface and the FutureTask class is analyzed.
Ok, that’s enough for today. I’m Glacier. See you next time