Java ThreadPoolExecutor: कोर पूल आकार को अद्यतन करना गतिशील रूप से आने वाले कार्यों को रुक-रुक कर अस्वीकार करता है

यहाँ एक परिदृश्य है कि ऐसा क्यों हो रहा है:

मेरे उदाहरण में मैं इसे छोटा करने के लिए minThreads = 0, maxThreads = 2 और queueCapacity = 2 का उपयोग करता हूं। पहला आदेश सबमिट हो जाता है, यह निष्पादन विधि में किया जाता है:

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);
}

इस आदेश के लिए addWorker(null, false) की तुलना में workQueue.offer(command) निष्पादित किया जाता है। वर्कर थ्रेड पहले इस कमांड को थ्रेड रन विधि में कतार से बाहर ले जाता है, इसलिए इस समय कतार में अभी भी एक कमांड है,

इस बार दूसरी कमांड सबमिट हो जाती है workQueue.offer(command) निष्पादित किया जाता है। अब कतार भर चुकी है

अब शेड्यूल्ड एक्ज़ीक्यूटर सर्विस resizeThreadPool मेथड को एक्जीक्यूट करती है जो setCorePoolSize को maxThreads के साथ कॉल करती है। यहाँ विधि है setCorePoolSize:

 public void setCorePoolSize(int corePoolSize) {
    if (corePoolSize < 0)
        throw new IllegalArgumentException();
    int delta = corePoolSize - this.corePoolSize;
    this.corePoolSize = corePoolSize;
    if (workerCountOf(ctl.get()) > corePoolSize)
        interruptIdleWorkers();
    else if (delta > 0) {
        // We don't really know how many new threads are "needed".
        // As a heuristic, prestart enough new workers (up to new
        // core size) to handle the current number of tasks in
        // queue, but stop if queue becomes empty while doing so.
        int k = Math.min(delta, workQueue.size());
        while (k-- > 0 && addWorker(null, true)) {
            if (workQueue.isEmpty())
                break;
        }
    }
}

यह विधि AddWorker (null, true) का उपयोग करके एक कार्यकर्ता को जोड़ती है। नहीं, 2 कार्यकर्ता कतारें चल रही हैं, अधिकतम और कतार भरी हुई है।

तीसरा आदेश सबमिट हो जाता है और विफल हो जाता है क्योंकि workQueue.offer(command) और addWorker(command, false) विफल हो जाता है, जिससे अपवाद होता है:

java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@24c22fe rejected from java.util.concurrent.ThreadPoolExecutor@cd1e646[Running, pool size = 2, active threads = 2, queued tasks = 2, 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 java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
at ThreadPoolResizeTest.executeOnce(ThreadPoolResizeTest.java:60)
at ThreadPoolResizeTest.runTest(ThreadPoolResizeTest.java:28)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:44)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:15)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:41)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:20)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:263)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:69)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:48)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:231)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:60)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:229)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:50)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:222)
at org.junit.runners.ParentRunner.run(ParentRunner.java:292)
at org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)

मुझे लगता है कि इस समस्या को हल करने के लिए आपको कतार की क्षमता को उस अधिकतम कमांड पर सेट करना चाहिए जिसे आप निष्पादित करना चाहते हैं।

सुनिश्चित नहीं है कि यह बग के रूप में योग्य है। यह व्यवहार तब होता है जब कतार भर जाने के बाद अतिरिक्त श्रमिक धागे बनाए जाते हैं लेकिन जावा डॉक्स में इस तरह का उल्लेख किया गया है कि कॉलर को अस्वीकार किए जा रहे कार्यों से निपटना पड़ता है।

जावा डॉक्स

नए धागे के लिए कारखाना। सभी धागे इस कारखाने का उपयोग करके बनाए गए हैं (विधि addWorker के माध्यम से)। सभी कॉलर्स को ऐडवर्कर के विफल होने के लिए तैयार रहना चाहिए, जो थ्रेड की संख्या को सीमित करने वाली प्रणाली या उपयोगकर्ता की नीति को प्रतिबिंबित कर सकता है। भले ही इसे एक त्रुटि के रूप में नहीं माना जाता है, थ्रेड्स बनाने में विफलता के परिणामस्वरूप नए कार्य अस्वीकृत हो सकते हैं या मौजूदा कार्य कतार में अटके रह सकते हैं।

जब आप कोर पूल आकार का आकार बदलते हैं, मान लें कि वृद्धि, अतिरिक्त कर्मचारी बनाए जाते हैं (addWorker विधि setCorePoolSize में) और अतिरिक्त कार्य बनाने के लिए कॉल (addWorker< /code> विधि से निष्पादित) को अस्वीकार कर दिया जाता है जब addWorker गलत (कार्यकर्ता जोड़ें अंतिम कोड स्निपेट) लौटाता है, क्योंकि पर्याप्त अतिरिक्त कर्मचारी पहले ही बनाए जा चुके होते हैं द्वारा setCorePoolSize लेकिन अभी तक नहीं चलाया गया ताकि अपडेट को कतार में प्रदर्शित किया जा सके।

public void setCorePoolSize(int corePoolSize) {
    ....
    int k = Math.min(delta, workQueue.size());
    while (k-- > 0 && addWorker(null, true)) {
        if (workQueue.isEmpty())
             break;
    }
}

public void execute(Runnable command) {
    ...
    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);
}

private boolean addWorker(Runnable firstTask, boolean core) {
....
   if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
     return false;             
}

public static class RetryRejectionPolicy implements RejectedExecutionHandler {
    public RetryRejectionPolicy () {}

    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
           while(true)
            if(e.getQueue().offer(r)) break;
        }
    }
}

ThreadPoolExecutor pool = new ThreadPoolExecutor(
      minThreads, maxThreads,
      0, TimeUnit.SECONDS,
      new LinkedBlockingQueue<Runnable>(queueCapacity),
      new ThreadPoolResizeTest.RetryRejectionPolicy()
 );