preface
- I read an article earlier, in which there was an interview question about multithreading.
- The question describes: What happens if a thread dies?
- When I didn’t read the follow-up of the article, I had several answers in my mind. Unfortunately, finally found no answer to the point, so intend to record -_-
The analysis process
- Start by defining a ThreadPoolExecutor thread pool. Add two more threads to it.
private final static ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(5.5.60, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
public static void main(String[] args) {
test();
}
private static void test(a) {
EXECUTOR.execute(() -> doIt("execute"));
EXECUTOR.submit(() -> doIt("submit"));
}
private static void doIt(String name) {
System.out.println(name + "Executed.");
throw new RuntimeException(name + "Abnormal.");
}
Copy the code- The result of execution is
Execute executes submit and Exception in Thread executes"pool-1-thread-1"Java. Lang. RuntimeException: execute the exception at com. Web. Test. Test20210226. Test1. DoIt (Test1. Java:26)
at com.web.test.test20210226.Test1.lambda$test$0(Test1.java:20)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Copy the code- Here’s our conclusion:
- Here, two threads are added to the thread pool. After throwing an exception, the threads do not affect each other.
- The execute() method of the thread pool throws an exception directly, whereas the submit() method here does not.
- The submit() method is usually followed by the Future.get () method to get the result of the thread’s execution. Now that the thread throws an exception, what will the result be?
private final static ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(5.5.60, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
public static void main(String[] args) {
test();
}
private static void test(a) { Future<? > future = EXECUTOR.submit(() -> doIt("submit"));
try {
future.get();
} catch(Exception e) { e.printStackTrace(); }}private static void doIt(String name) {
System.out.println(name + "Executed.");
throw new RuntimeException(name + "Abnormal.");
}
Copy the code- Execution result:
Submit to perform the Java. Util. Concurrent. ExecutionException: Java. Lang. RuntimeException: Submit the exception at Java. Util. Concurrent. FutureTask. Report (FutureTask. Java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at com.web.test.test20210226.Test1.test(Test1.java:23)
at com.web.test.test20210226.Test1.main(Test1.java:17) under Caused by: Java. Lang. RuntimeException: submit the exception at com. Web. Test. Test20210226. Test1. DoIt (Test1. Java:31)
at com.web.test.test20210226.Test1.lambda$test$0(Test1.java:21)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Copy the code- Here, too, an exception message is thrown after the future.get() method is called. So it can be concluded that:
- The thread pool’s submit() method does not throw an exception directly, as it would if the Future.get () method were called.
- According to the print information, there are printed at the Java. Util. Concurrent. ThreadPoolExecutor. RunWorker (ThreadPoolExecutor. Java: 1149). Click in to see the ThreadPoolExecutor source code.
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while(task ! =null|| (task = getTask()) ! =null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && ! wt.isInterrupted()) wt.interrupt();try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run(); // Throw an exception
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally{ afterExecute(task, thrown); }}finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// Thread pool worker recycling mechanism implementation.processWorkerExit(w, completedAbruptly); }}Copy the code- You can see that after the task.run() method exception, the catch statement directly throws the exception message.
- Then look at the processWorkerExit(w, completedAbruptly) method. This method is to realize the recycling mechanism of worker when the thread is interrupted abnormally.
/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
// Remove the thread that was previously interrupted abnormally
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if(! completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0&&! workQueue.isEmpty()) min =1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// Add another thread to the queue
addWorker(null.false); }}Copy the code
- So here’s the conclusion: When a thread throw exception occurs, the thread is not reclaimed to the thread pool, but directly removed, and a new thread is created into the thread pool.
- Take a look at the submit(Runnable Task) method to execute the source code. The parameters passed in are encapsulated as a FutureTask object and are also executed by the execute() method.
/ * * *@throws RejectedExecutionException {@inheritDoc}
* @throws NullPointerException {@inheritDoc} * /
publicFuture<? > submit(Runnable task) {if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
returnftask; } -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -/**
* Returns a {@code RunnableFuture} for the given runnable and default
* value.
*
* @param runnable the runnable task being wrapped
* @param value the default value for the returned future
* @param <T> the type of the given value
* @return a {@code RunnableFuture} which, when run, will run the
* underlying runnable and which, as a {@code Future}, will yield
* the given value as its result and provide for cancellation of
* the underlying task
* @since1.6 * /
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
Copy the code- Look at the source code for the run() method executed by the FutureTask object
/**
* 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)) {
// t is the exception information stored in the run() method
outcome = t;
// Assign EXCEPTIONAL (3) to state
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion(); }}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;
// Save the exception information and change the value of the variable state
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); }} -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --/** * 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; // Create an initial state
private static final int COMPLETING = 1; / / run
private static final int NORMAL = 2; / / normal
private static final int EXCEPTIONAL = 3; / / exception
private static final int CANCELLED = 4; / / cancel
private static final int INTERRUPTING = 5; / / the interrupt
private static final int INTERRUPTED = 6; / / is interrupted
Copy the code- In catch statements, instead of throwing the exception information directly, the setException(ex) method is called to save the exception information. And assign the variable state to EXCEPTIONAL
- Take a look at the source execution process of the fufuretask.get() method.
/ * * *@throws CancellationException {@inheritDoc} * /
public V get(a) throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
// If the state of the variable is greater than 1, return state. So this returns the modified value 3 from the setException() method
s = awaitDone(false.0L);
returnreport(s); } -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --/**
* 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 an exception when the value of state is 3
throw new ExecutionException((Throwable)x);
}
Copy the code- Throw an ExecutionException and previously saved exception information directly in the report() method.
The last
- The conclusions are as follows:
- If the thread is executed using the execute method, the exception will be thrown directly. If the thread is submitted using the submit method, the exception will not be printed directly. Instead, the future.get() method will be called to print the exception.
- When a thread throws an exception, other threads will not be affected.
- When a thread throws an exception, the thread is removed from the pool, and a new thread is created and put into the pool.
- Study with an open mind and make progress together
- Links to articles