मैं एक ऐसे मुद्दे में भाग रहा हूं जहां अगर मैं पूल के निर्माण के बाद एक ThreadPoolExecutor के कोर पूल आकार को एक अलग संख्या में आकार देने का प्रयास करता हूं, तो रुक-रुक कर, कुछ कार्यों को RejectedExecutionException के साथ अस्वीकार कर दिया जाता है, भले ही मैं कभी भी queueSize + maxPoolSize से अधिक कार्यों को सबमिट नहीं करता।

जिस समस्या को मैं हल करने का प्रयास कर रहा हूं वह ThreadPoolExecutor का विस्तार करना है जो थ्रेड पूल की कतार में बैठे लंबित निष्पादन के आधार पर अपने मूल धागे का आकार बदलता है। मुझे इसकी आवश्यकता है क्योंकि डिफ़ॉल्ट रूप से एक ThreadPoolExecutor एक नया Thread तभी बनाएगा जब कतार भरी हो।

यहाँ एक छोटा स्व-निहित शुद्ध जावा 8 प्रोग्राम है जो समस्या को प्रदर्शित करता है।

import static java.lang.Math.max;
import static java.lang.Math.min;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolResizeTest {

    public static void main(String[] args) throws Exception {
        // increase the number of iterations if unable to reproduce
        // for me 100 iterations have been enough
        int numberOfExecutions = 100;

        for (int i = 1; i <= numberOfExecutions; i++) {
            executeOnce();
        }
    }

    private static void executeOnce() throws Exception {
        int minThreads = 1;
        int maxThreads = 5;
        int queueCapacity = 10;

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

        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> resizeThreadPool(pool, minThreads, maxThreads),
                0, 10, TimeUnit.MILLISECONDS);
        CompletableFuture<Void> taskBlocker = new CompletableFuture<>();

        try {
            int totalTasksToSubmit = queueCapacity + maxThreads;

            for (int i = 1; i <= totalTasksToSubmit; i++) {
                // following line sometimes throws a RejectedExecutionException
                pool.submit(() -> {
                    // block the thread and prevent it from completing the task
                    taskBlocker.join();
                });
                // Thread.sleep(10); //enabling even a small sleep makes the problem go away
            }
        } finally {
            taskBlocker.complete(null);
            scheduler.shutdown();
            pool.shutdown();
        }
    }

    /**
     * Resize the thread pool if the number of pending tasks are non-zero.
     */
    private static void resizeThreadPool(ThreadPoolExecutor pool, int minThreads, int maxThreads) {
        int pendingExecutions = pool.getQueue().size();
        int approximateRunningExecutions = pool.getActiveCount();

        /*
         * New core thread count should be the sum of pending and currently executing tasks
         * with an upper bound of maxThreads and a lower bound of minThreads.
         */
        int newThreadCount = min(maxThreads, max(minThreads, pendingExecutions + approximateRunningExecutions));

        pool.setCorePoolSize(newThreadCount);
        pool.prestartAllCoreThreads();
    }
}

पूल को कभी भी अस्वीकृत निष्पादन अपवाद क्यों फेंकना चाहिए यदि मैं कभी भी कतार क्षमता + अधिकतम थ्रेड सबमिट नहीं करता हूं। मैं अधिकतम धागे को कभी नहीं बदल रहा हूं इसलिए ThreadPoolExecutor की परिभाषा के अनुसार, इसे या तो कार्य को थ्रेड या कतार में समायोजित करना चाहिए।

बेशक, अगर मैं पूल का आकार कभी नहीं बदलता, तो थ्रेड पूल कभी भी किसी सबमिशन को अस्वीकार नहीं करता है। इसे डिबग करना भी कठिन है क्योंकि सबमिशन में किसी भी प्रकार की देरी जोड़ने से समस्या दूर हो जाती है।

RejectedExecutionException को ठीक करने के तरीके पर कोई संकेतक?

14
Swaranga Sarma 11 मार्च 2020, 08:47
क्यों न किसी मौजूदा को लपेटकर ExecutorService का अपना कार्यान्वयन प्रदान किया जाए, जो आकार बदलने के कारण सबमिट करने में विफल कार्यों को फिर से सबमिट करता है?
 – 
daniu
13 मार्च 2020, 09:07
1
यह एक उपाय है। प्रश्नों का मुद्दा यह है कि पूल को कभी भी अस्वीकृत निष्पादन अपवाद क्यों फेंकना चाहिए यदि मैं कभी भी कतार क्षमता + अधिकतम थ्रेड सबमिट नहीं करता हूं। मैं अधिकतम धागे को कभी नहीं बदल रहा हूं इसलिए ThreadPoolExecutor की परिभाषा के अनुसार, इसे या तो कार्य को थ्रेड या कतार में समायोजित करना चाहिए।
 – 
Swaranga Sarma
13 मार्च 2020, 09:09
ठीक है, ऐसा लगता है कि मैंने आपके प्रश्न को गलत समझा है। यह क्या है? क्या आप जानना चाहते हैं कि व्यवहार क्यों होता है या आप इसे कैसे प्राप्त करते हैं जिससे आपके लिए समस्याएं पैदा होती हैं?
 – 
daniu
13 मार्च 2020, 10:20
हां, मेरे कार्यान्वयन को एक निष्पादक सेवा में बदलना संभव नहीं है क्योंकि बहुत सारे कोड ThreadPoolExecutor को संदर्भित करते हैं। तो अगर मैं अभी भी एक आकार बदलने योग्य ThreadPoolExecutor रखना चाहता हूं तो मुझे यह जानने की जरूरत है कि मैं इसे कैसे ठीक कर सकता हूं। ऐसा कुछ करने का सही तरीका है ThreadPoolExecutor का विस्तार करना और इसके कुछ संरक्षित चरों तक पहुंच प्राप्त करना और सुपर क्लास द्वारा साझा किए गए लॉक पर एक सिंक्रनाइज़ ब्लॉक के भीतर पूल आकार को अपडेट करना।
 – 
Swaranga Sarma
13 मार्च 2020, 10:39
ThreadPoolExecutor का विस्तार करना शायद एक बुरा विचार है, और क्या आपको इस मामले में भी मौजूदा कोड को बदलने की आवश्यकता नहीं होगी? यह सबसे अच्छा होगा कि आप कुछ उदाहरण प्रदान करें कि आपका वास्तविक कोड निष्पादक तक कैसे पहुंचता है। मुझे आश्चर्य होगा अगर यह ThreadPoolExecutor के लिए विशिष्ट कई विधियों का उपयोग करता है (अर्थात ExecutorService में नहीं)।
 – 
daniu
13 मार्च 2020, 10:57

2 जवाब

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

मेरे उदाहरण में मैं इसे छोटा करने के लिए 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)

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

5
Thomas Krieger 14 मार्च 2020, 14:34
सही। मैं कोड को अपनी कक्षा में कॉपी करके और लॉगर्स जोड़कर पुन: पेश करने में सक्षम था। असल में, जब कतार भर जाती है, और मैं एक नया कार्य सबमिट करता हूं, तो यह एक नया कार्यकर्ता बनाने का प्रयास करेगा। इस बीच अगर उस समय, मेरा रिसाइज़र भी सेटकोरपूलसाइज़ को 2 पर कॉल करता है, तो यह एक नया वर्कर भी बनाता है। इस बिंदु पर, दो श्रमिक जोड़े जाने के लिए प्रतिस्पर्धा कर रहे हैं, लेकिन वे दोनों नहीं हो सकते क्योंकि यह अधिकतम-पूल-आकार की बाधा का उल्लंघन करेगा इसलिए नया कार्य सबमिशन अस्वीकार कर दिया जाता है। मुझे लगता है कि यह एक दौड़ की स्थिति है और मैंने ओपनजेडीके को एक बग रिपोर्ट दर्ज की है। आइए देखते हैं। लेकिन आपने मेरे प्रश्न का उत्तर दिया ताकि आपको बक्षीस मिले। धन्यवाद।
 – 
Swaranga Sarma
20 मार्च 2020, 20:27

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

जावा डॉक्स

नए धागे के लिए कारखाना। सभी धागे इस कारखाने का उपयोग करके बनाए गए हैं (विधि 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()
 );

यह भी ध्यान दें कि शटडाउन का आपका उपयोग सही नहीं है क्योंकि यह सबमिट किए गए कार्य के निष्पादन को पूरा करने की प्रतीक्षा नहीं करेगा बल्कि इसके बजाय awaitTermination का उपयोग करेगा।

2
s7vr 17 मार्च 2020, 16:34
मुझे लगता है कि जावाडॉक के अनुसार शटडाउन पहले से सबमिट किए गए कार्यों की प्रतीक्षा करता है: शटडाउन() एक व्यवस्थित शटडाउन शुरू करता है जिसमें पहले सबमिट किए गए कार्यों को निष्पादित किया जाता है, लेकिन कोई नया कार्य स्वीकार नहीं किया जाएगा।
 – 
Thomas Krieger
16 मार्च 2020, 19:14
- यह पहले से सबमिट किए गए कार्यों को निष्पादित करेगा लेकिन उनके समाप्त होने की प्रतीक्षा नहीं करेगा - डॉक्स docs.oracle.com/javase/7/docs/api/java/util/concurrent/… - यह विधि निष्पादन को पूरा करने के लिए पहले सबमिट किए गए कार्यों की प्रतीक्षा नहीं करती है . ऐसा करने के लिए प्रतीक्षा समाप्ति का उपयोग करें।
 – 
s7vr
16 मार्च 2020, 21:05