Ist das hinzufügen von Aufgaben zu BlockingQueue von ThreadPoolExecutor ratsam?

Wenn es nach mir ginge, würde ich lieber mit Executor#execute() über Queue#offer() einfach, denn ich bin mit alles aus java.util.concurrent schon.

Deine Frage ist gut, und es hat mein Interesse geweckt, so nahm ich einen Blick auf die Quelle für ThreadPoolExecutor#execute():

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        if (runState == RUNNING && workQueue.offer(command)) {
            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }
        else if (!addIfUnderMaximumPoolSize(command))
            reject(command); //is shutdown or saturated
    }
}

Können wir sehen, dass führen Sie selbst nennt offer() auf der Warteschlange arbeiten, aber nicht, bevor Sie einige schöne, leckere pool Manipulationen, wenn nötig. Aus diesem Grund, würde ich denken, es wäre ratsam, zu verwenden execute(); nicht verwenden können (obwohl ich nicht weiß, für bestimmte) Ursache der pool zu betreiben in einem nicht-optimalen Weg. Aber ich glaube nicht, dass mit offer() wird Pause der Testamentsvollstrecker - wie es aussieht-Aufgaben gezogen werden aus der Warteschlange mit den folgenden (auch von ThreadPoolExecutor):

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            if (state == SHUTDOWN)  //Help drain queue
                r = workQueue.poll();
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            else
                r = workQueue.take();
            if (r != null)
                return r;
            if (workerCanExit()) {
                if (runState >= SHUTDOWN) //Wake up others
                    interruptIdleWorkers();
                return null;
            }
            //Else retry
        } catch (InterruptedException ie) {
            //On interruption, re-check runState
        }
    }
}

Diese getTask() Methode wird nur aufgerufen innerhalb einer Schleife, so dass, wenn der Testamentsvollstrecker ist nicht heruntergefahren, es würde blockieren, bis eine neue Aufgabe gegeben wurde, um die Warteschlange (unabhängig davon, von wo es kam).

Hinweis: obwohl ich gebucht habe, code-snippets von Quelle hier, wir können nicht auf Sie verlassen für eine definitive Antwort - wir sollten nur Codierung auf die API. Wir wissen nicht, wie die Umsetzung der execute() im Laufe der Zeit verändern.

Kann man eigentlich konfigurieren Verhalten des Pools, wenn die Warteschlange voll ist, werden durch die Angabe einer RejectedExecutionHandler bei der Instanziierung. ThreadPoolExecutor definiert vier Strategien, wie innere Klassen, einschließlich AbortPolicy, DiscardOldestPolicy, DiscardPolicy sowie mein persönlicher Favorit, CallerRunsPolicy, läuft der neue job in der controlling-thread.

Beispiel:

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
        nproc, //core size
        nproc, //max size
        60, //idle timeout
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(4096, true), //Fairness = true guarantees FIFO
        new ThreadPoolExecutor.CallerRunsPolicy() ); //If we have to reject a task, run it in the calling thread.

Das gewünschte Verhalten in der Frage gewonnen werden können, mit so etwas wie:

public class BlockingPolicy implements RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        executor.getQueue.put(r); //Self contained, no queue reference needed.
    }

Irgendwann die Warteschlange muss zugegriffen werden. Der beste Ort, um dies zu tun, eine eigenständige RejectedExecutionHandler, die speichert alle code-Duplizierung oder potenial Fehler, die sich aus direkten manipulation der Warteschlange auf den Geltungsbereich des pool-Objekts. Beachten Sie, dass die Prozeduren enthalten in ThreadPoolExecutor selbst getQueue().

Falls erforderlich, können wir auch ein Parkplatz trennt die Haupt-Verarbeitung von abgelehnten Aufgaben -

    final CountDownLatch taskCounter = new CountDownLatch(TASKCOUNT);
    final List<Runnable> taskParking = new LinkedList<Runnable>();
    BlockingQueue<Runnable> taskPool = new ArrayBlockingQueue<Runnable>(1);
    RejectedExecutionHandler rejectionHandler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            System.err.println(Thread.currentThread().getName() + " -->rejection reported - adding to parking lot " + r);
            taskCounter.countDown();
            taskParking.add(r);
        }
    };
    ThreadPoolExecutor threadPoolExecutor =  new ThreadPoolExecutor(5, 10, 1000, TimeUnit.SECONDS, taskPool, rejectionHandler);
    for(int i=0 ; i<TASKCOUNT; i++){
        //main 
        threadPoolExecutor.submit(getRandomTask());
    }
    taskCounter.await(TASKCOUNT * 5 , TimeUnit.SECONDS);
    System.out.println("Checking the parking lot..." + taskParking);
    while(taskParking.size() > 0){
        Runnable r = taskParking.remove(0);
        System.out.println("Running from parking lot..." + r);
        if(taskParking.size() > LIMIT){
          waitForSometime(...);
        }
        threadPoolExecutor.submit(r);
    }
    threadPoolExecutor.shutdown();