7

ThreadPoolExecutorを使用すると、コンストラクターで渡されるか、メソッドを使用しRunnableて実行される s を送信できると思いました。 また、私の理解では、タスクが利用可能であれば実行されます。 私が理解していないのは次のとおりです。 BlockingQueueexecute

public class MyThreadPoolExecutor {  

    private static ThreadPoolExecutor executor;  

    public MyThreadPoolExecutor(int min, int max, int idleTime, BlockingQueue<Runnable> queue){  
        executor = new ThreadPoolExecutor(min, max, 10, TimeUnit.MINUTES, queue);   
        //executor.prestartAllCoreThreads();  
    }  

    public static void main(String[] main){
        BlockingQueue<Runnable> q = new LinkedBlockingQueue<Runnable>();
        final String[] names = {"A","B","C","D","E","F"};  
        for(int i = 0; i < names.length; i++){  
            final int j = i;  
            q.add(new Runnable() {  

                @Override  
                public void run() {  
                    System.out.println("Hi "+ names[j]);  

                }  
            });         
        }  
        new MyThreadPoolExecutor(10, 20, 1, q);   
        try {  
            TimeUnit.SECONDS.sleep(5);  
        } catch (InterruptedException e) {  
            // TODO Auto-generated catch block  
            e.printStackTrace();  
        }  
        /*executor.execute(new Runnable() {  

            @Override  
            public void run() {  

                System.out.println("++++++++++++++");  

            }   
        });  */
        for(int i = 0; i < 100; i++){  
            final int j = i;  
            q.add(new Runnable() {   

                @Override  
                public void run() {  
                    System.out.println("Hi "+ j);  

                }  
            });  
        }   


    }  


}

executor.prestartAllCoreThreads();このコードは、コンストラクターでのコメントを解除するかexecute、出力する実行可能ファイルを呼び出さない限り、何もしませんSystem.out.println("++++++++++++++");(これもコメントアウトされています)。

なんで？
引用（私の強調）：

デフォルトでは、コアスレッドも最初に作成され、新しいタスクが到着したときにのみ開始されますが、メソッド prestartCoreThread() または prestartAllCoreThreads() を使用して動的にオーバーライドできます。空でないキューを使用してプールを構築する場合は、おそらくスレッドを事前に開始する必要があります。

Ok。だから私のキューは空ではありません。しかしexecutor、私はを作成し、作成してから、sleep新しいRunnableをキューに追加します (100 へのループ内)。
このループは とカウントされnew tasks arriveませんか?
なぜそれが機能しないので、明示的に呼び出すか、prestartまたは明示的に呼び出す必要がありますexecuteか?

4

2 に答える 2

10

ワーカー スレッドは、実行によってタスクが到着すると生成されます。これらは、基になるワーク キューとやり取りするスレッドです。空でないワーク キューから開始する場合は、ワーカーを事前に開始する必要があります。OpenJDK 7 での実装を参照してください。

繰り返しますが、work queue と対話するのはワーカーです。経由で渡されたときにオンデマンドでのみ生成されexecuteます。(またはその上のレイヤー、たとえばinvokeAll、など) それらが開始されていない場合、開始されたワーカーがないsubmitため、キューをチェックするものが何もないため、キューに追加する作業の量は問題になりません。

ThreadPoolExecutor必要になるまで、またはメソッドprestartAllCoreThreadsおよびprestartCoreThreadによってワーカー スレッドの作成を先取りするまで、ワーカー スレッドを生成しません。ワーカーが開始されていない場合、キュー内の作業が実行されることはありません。

初期作業を追加する理由は、単一のコア ワーカー スレッドexecuteの作成を強制するためです。これにより、キューから作業の処理を開始できます。同様の動作を呼び出して受信することもできます。すべてのワーカーを開始したい場合は、その数のタスクを呼び出すか、 経由で送信する必要があります。prestartCoreThreadprestartAllCoreThreadsexecute

以下のコードを参照してくださいexecute。

/**
 * Executes the given task sometime in the future.  The task
 * may execute in a new thread or in an existing pooled thread.
 *
 * If the task cannot be submitted for execution, either because this
 * executor has been shutdown or because its capacity has been reached,
 * the task is handled by the current {@code RejectedExecutionHandler}.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 *         {@code RejectedExecutionHandler}, if the task
 *         cannot be accepted for execution
 * @throws NullPointerException if {@code command} is null
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}
于 2012-08-05T21:23:08.473 に答える
5

BlockingQueue は魔法のスレッド ディスパッチャーではありません。Runnable オブジェクトをキューに送信し、それらのタスクを消費する実行中のスレッドがない場合、それらはもちろん実行されません。一方、execute メソッドは、必要に応じて、スレッド プールの構成に従ってスレッドを自動的にディスパッチします。すべてのコア スレッドを事前に開始すると、キューからタスクを消費するスレッドが存在します。

于 2012-08-05T21:43:44.757 に答える