スレッドプールの閉じ方はいくつかありますが、それぞれの違いは何ですか.

61749 ワード

一、概説Javaが提供するインタフェースjava.util.concurrent.ExecutorServiceは、タスクをバックグラウンドで実行できる非同期実行メカニズムである.インスタンスはスレッドプールのようなもので、タスクを統一的に管理できます.
二、研究Javaによって提供されるExecutorServiceに対するクローズ方法は、shutdown()メソッドを呼び出す方法と、shutdownNow()メソッドを呼び出す方法の2つである.この二つには違いがある.
次の内容は、ソースコード内のコメントから抜粋します.

// shutdown()
Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
Invocation has no additional effect if already shut down.
This method does not wait for previously submitted tasks to complete execution.  Use awaitTermination to do that.

// shutdownNow()
Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution.
This method does not wait for actively executing tasks to terminate.  Use awaitTermination to do that.
There are no guarantees beyond best-effort attempts to stop processing actively executing tasks.  For example, typical implementations will cancel via interrupt, so any task that fails to respond to interrupts may never terminate.

2つの英語はどういう意味ですか.簡単にまとめてみます.
shutdown:1、呼び出し後、スレッドプールにスレッドを追加し続けることは許可されません.2、スレッドプールの状態がSHUTDOWN状態になる;3、shutdown()メソッドを呼び出す前にExecutorSrviceにコミットされたすべてのタスクが実行されます.4、すべてのスレッドが現在のタスクの実行を終了すると、ExecutorServiceは本当に閉じます.
shutdownNow():1、この方法はまだ実行されていないtaskのリストを返す.2、スレッドプールの状態がSTOP状態になる;3、起動待ちのすべてのタスクをブロックし、現在実行中のタスクを停止します.
簡単に言えば、shutdown()呼び出し後、submitの新しいtaskは使用できません.submitのshutdownNow()呼び出しが継続された後、現在実行中のtaskを停止し、まだ実行されていないtaskのlistに戻ります.
三、ソース分析
ここではJDK 1を使っています.8,まずThreadPoolExecutorのshutDown()に入る方法:

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

private void checkShutdownAccess() {
   SecurityManager security = System.getSecurityManager();
   if (security != null) {
       security.checkPermission(shutdownPerm);
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           for (Worker w : workers)
               security.checkAccess(w.thread);
       } finally {
           mainLock.unlock();
       }
   }
}

private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}

void onShutdown() {}

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

shutDownNow()の方法に入ってみてください.

public List<Runnable> shutdownNow() {
   List<Runnable> tasks;
   final ReentrantLock mainLock = this.mainLock;
   mainLock.lock();
   try {
       checkShutdownAccess();
       advanceRunState(STOP);
       interruptWorkers();
       tasks = drainQueue();
   } finally {
       mainLock.unlock();
   }
   tryTerminate();
   return tasks;
}

private void checkShutdownAccess() {
   SecurityManager security = System.getSecurityManager();
    if (security != null) {
        security.checkPermission(shutdownPerm);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                security.checkAccess(w.thread);
        } finally {
            mainLock.unlock();
        }
    }
}

private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

private List<Runnable> drainQueue() {
    BlockingQueue<Runnable> q = workQueue;
    ArrayList<Runnable> taskList = new ArrayList<Runnable>();
    q.drainTo(taskList);
    if (!q.isEmpty()) {
        for (Runnable r : q.toArray(new Runnable[0])) {
            if (q.remove(r))
                taskList.add(r);
        }
    }
    return taskList;
}

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

四、実戦
1、Demo1

package com.concurrent.executorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author riemann
 * @date 2019/07/28 23:41
 */
public class ExecutorServiceDemo1 {

    static Runnable run = () -> {
        try {
            Thread.sleep(5000);
            System.out.println("thread finish");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    };

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(2);
        service.execute(run);
        service.shutdown();
        service.execute(run);
    }

}

出力結果:

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.concurrent.executorService.ExecutorServiceDemo1$$Lambda$1/1854731462@312b1dae rejected from java.util.concurrent.ThreadPoolExecutor@7530d0a[Shutting down, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
	at com.concurrent.executorService.ExecutorServiceDemo1.main(ExecutorServiceDemo1.java:25)
thread finish

shutdown()が呼び出されると、タスクの追加は続行できません.そうしないと、例外RejectedExecutionExceptionが放出されます.スレッドプールは、実行中のタスクが終了すると本当に終了します.
2、Demo2

package com.concurrent.executorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author riemann
 * @date 2019/07/29 0:03
 */
public class ExecutorServiceDemo2 {

    static Runnable run = () -> {
        try {
            Thread.sleep(5000);
            System.out.println("thread finish");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    };

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(2);
        service.execute(run);
        service.shutdownNow();
    }

}

出力結果:

java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at com.concurrent.executorService.ExecutorServiceDemo2.lambda$static$0(ExecutorServiceDemo2.java:14)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)

shutdownNow()を使用して、sleep/wait/実行タイミングロックなどがスレッドにある場合は、実行中のスレッドを直接終了し、interrupt異常を投げ出す.その内部はThread.interrupt()によって実現されるからである.しかし、この方法には強い限界がある.sleepなどの方法が実行されていなければ、スレッドを終了することはできないからです.
3、Demo3

package com.concurrent.executorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author riemann
 * @date 2019/07/29 0:05
 */
public class ExecutorServiceDemo3 {

    static Runnable run = () -> {
        long num = 0;
        boolean flag = true;
        while (flag) {
            num += 1;
            if (num == Long.MAX_VALUE) {
                flag = false;
            }
        }
    };

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(1);
        service.execute(run);
        service.shutdownNow();
    }

}

多くのコードでは、例えば、ループタグflagを使用して、ある条件が満たされるまで、ループタグfalseを設定するまで、いくつかの時間のかかる計算タスクをループで実行する場合があります.Demo 3コードに示すように(ループ待ちの場合)、shutdownNow()はスレッドを終了できません.このような場合は、Demo 4のような方法を使用することができます.
4、Demo4

package com.concurrent.executorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author riemann
 * @date 2019/07/29 0:12
 */
public class ExecutorServiceDemo4 {

    static Runnable run = () -> {
        long num = 0;
        while (true && !Thread.currentThread().isInterrupted()) {
            num += 1;
        }
        System.out.println(num);
    };

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(1);
        service.execute(run);
        service.shutdownNow();
    }

}

出力結果:

ループ待ちの場合、変数Thread.currentThread().isInterrupted()を1つの判定条件として導入することができる.isInterrupted()メソッドは、現在のスレッドがinterruptされているかどうかを返します.shutdownNow()の内部実装は、実際にはinterruptによってスレッドを終了するので、shutdownNow()が呼び出されると、isInterrupted()はtrueを返す.このときループから飛び出して待つことができます.しかし、これは最も優雅な解決策ではなく、具体的にはDemo 5を参照することができる.
5、Demo5

package com.concurrent.executorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author riemann
 * @date 2019/07/29 0:17
 */
public class ExecutorServiceDemo5 {

    static Runnable run = () -> {
        long num = 0;
        boolean flag = true;
        while (flag && !Thread.currentThread().isInterrupted()) {
            num += 1;
            if (num == Long.MAX_VALUE) {
                flag = false;
            }
        }
        System.out.println(num);
    };

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(1);
        service.execute(run);
        service.shutdown();
        try {
            if (!service.awaitTermination(2, TimeUnit.SECONDS)) {
                service.shutdownNow();
            }
        } catch (InterruptedException e) {
            service.shutdownNow();
        }
    }

}

出力結果:

999032162

ここです.shutdown()を呼び出してスレッドプールのステータスをSHUTDOWNに変更し、スレッドプールはスレッドの追加を継続することを許可せず、実行中のスレッドの戻りを待つ.awaitTerminationを呼び出してタイミングタスクを設定し、コード内の意味は2 s後にスレッドプール内のスレッドがすべて実行済みかどうかを検出し(先生が学生に「最後に2 s時間で宿題を書き終える」と言ったように)、実行が完了しなければshutdownNow()メソッドを呼び出す.

Ubuntu で Digdag サーバを作るまでの手順

Treasure Dataのtdコマンドの始め方