Nie można utworzyć buforowanej puli wątków z ograniczeniem rozmiaru?

Wydaje się, że niemożliwe jest utworzenie buforowanej puli wątków z ograniczeniem liczby wątków, które może ona utworzyć.

Oto jak static Executors.newCachedThreadPool jest zaimplementowane w standardowej bibliotece Java:

 public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

Tak więc, używając tego szablonu do utworzenia puli wątków w pamięci podręcznej o stałym rozmiarze:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronusQueue<Runable>());

Teraz, jeśli użyjesz tego i prześlesz 3 zadania, wszystko będzie dobrze. Przesłanie dalszych zadań spowoduje odrzucenie wyjątków wykonania.

Próbuję tego:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runable>());

Spowoduje, że wszystkie wątki będą wykonywane sekwencyjnie. Oznacza to, że pula wątków nigdy nie utworzy więcej niż jednego wątku do obsługi zadań.

To jest błąd w metodzie wykonywania programu ThreadPoolExecutor? A może jest to zamierzone? Czy jest inny sposób?

Edycja: chcę coś dokładnie takiego jak buforowana pula wątków (tworzy wątki na żądanie, a następnie je zabija po pewnym czasie), ale z ograniczeniem liczby wątków, które może utworzyć i możliwością kontynuowania kolejkowania dodatkowych zadań, gdy ma osiągnął limit wątków. Zgodnie z odpowiedzią sjlee jest to niemożliwe. Patrząc na metodę execute () ThreadPoolExecutor, jest to rzeczywiście niemożliwe. Musiałbym podklasować ThreadPoolExecutor i przesłonić execute (), podobnie jak robi to SwingWorker, ale to, co robi SwingWorker w swoim execute (), to kompletny hack.

ThreadPoolExecutor ma następujące kilka kluczowych zachowań, a Twoje problemy można wyjaśnić za pomocą tych zachowań.

Po przesłaniu zadań

1. Jeśli pula wątków nie osiągnęła rozmiaru rdzenia, tworzy nowe wątki.
2. Jeśli osiągnięto rozmiar rdzenia i nie ma bezczynnych wątków, kolejkuje zadania.
3. Jeśli osiągnięto rozmiar rdzenia, nie ma bezczynnych wątków, a kolejka zapełni się, tworzy nowe wątki (aż osiągnie maksymalny rozmiar).
4. Jeśli osiągnięto maksymalny rozmiar, nie ma żadnych bezczynnych wątków, a kolejka zapełni się, rozpoczyna się zasada odrzucania.

W pierwszym przykładzie zwróć uwagę, że SynchronousQueue ma zasadniczo rozmiar 0. W związku z tym w momencie osiągnięcia maksymalnego rozmiaru (3) zaczyna obowiązywać zasada odrzucania (# 4).

W drugim przykładzie wybraną kolejką jest LinkedBlockingQueue o nieograniczonym rozmiarze. Dlatego utkniesz z zachowaniem # 2.

Nie możesz naprawdę majstrować przy typie buforowanym lub typie stałym, ponieważ ich zachowanie jest prawie całkowicie określone.

Jeśli chcesz mieć ograniczoną i dynamiczną pulę wątków, musisz użyć dodatniego rozmiaru rdzenia i maksymalnego rozmiaru w połączeniu z kolejką o skończonym rozmiarze. Na przykład,

new ThreadPoolExecutor(10, // core size
    50, // max size
    10*60, // idle timeout
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<Runnable>(20)); // queue with a size

Dodatek : jest to dość stara odpowiedź i wydaje się, że JDK zmienił swoje zachowanie, jeśli chodzi o rozmiar rdzenia równy 0. Od JDK 1.6, jeśli rozmiar rdzenia wynosi 0, a pula nie zawiera żadnych wątków, ThreadPoolExecutor doda wątku, aby wykonać to zadanie. Dlatego rozmiar rdzenia równy 0 jest wyjątkiem od powyższej reguły. Dziękuję Steve za zwrócenie mi na to uwagi.

O ile czegoś nie przeoczyłem, rozwiązanie pierwotnego pytania jest proste. Poniższy kod implementuje żądane zachowanie zgodnie z opisem w oryginalnym plakacie. Pojawi się do 5 wątków, aby pracować w nieograniczonej kolejce, a bezczynne wątki zostaną zakończone po 60 sekundach.

tp = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>());
tp.allowCoreThreadTimeOut(true);

Miał ten sam problem. Ponieważ żadna inna odpowiedź nie łączy wszystkich problemów, dodaję moje:

Jest teraz wyraźnie napisane dokumentacji : jeśli używasz kolejki, która nie blokuje ( LinkedBlockingQueue) maksymalnej liczby wątków, ustawienie nie ma wpływu, używane są tylko podstawowe wątki.

więc:

public class MyExecutor extends ThreadPoolExecutor {

    public MyExecutor() {
        super(4, 4, 5,TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        allowCoreThreadTimeOut(true);
    }

    public void setThreads(int n){
        setMaximumPoolSize(Math.max(1, n));
        setCorePoolSize(Math.max(1, n));
    }

}

Ten wykonawca ma:

1. Brak koncepcji maksymalnej liczby wątków, ponieważ używamy nieograniczonej kolejki. Jest to dobra rzecz, ponieważ taka kolejka może spowodować, że moduł wykonawczy utworzy ogromną liczbę dodatkowych wątków niebędących rdzeniami, jeśli będzie postępować zgodnie ze swoją zwykłą polityką.

2. Kolejka o maksymalnym rozmiarze Integer.MAX_VALUE. Submit()wyrzuci, RejectedExecutionExceptionjeśli liczba oczekujących zadań przekroczyInteger.MAX_VALUE . Nie jestem pewien, czy najpierw zabraknie nam pamięci, bo tak się stanie.

3. Możliwe 4 wątki rdzeniowe. Wątki bezczynności są automatycznie zamykane, jeśli są bezczynne przez 5 sekund.Tak więc, tak, wątki ściśle na żądanie.Liczba może być zmieniana za pomocąsetThreads() metody.

4. Zapewnia, że ​​minimalna liczba głównych wątków nigdy nie jest mniejsza niż jeden, w przeciwnym razie submit()odrzuci każde zadanie. Ponieważ wątki podstawowe muszą być> = max wątkami, metoda setThreads()ustawia również maksymalną liczbę wątków, chociaż ustawienie maksymalnego wątku jest bezużyteczne dla nieograniczonej kolejki.

W pierwszym przykładzie kolejne zadania są odrzucane, ponieważ AbortPolicyjest to ustawienie domyślne RejectedExecutionHandler. ThreadPoolExecutor zawiera następujące zasady, które można zmienić za pomocą setRejectedExecutionHandlermetody:

CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy

Wygląda na to, że chcesz buforować pulę wątków z CallerRunsPolicy.

Żadna z odpowiedzi tutaj nie rozwiązała mojego problemu, który dotyczył tworzenia ograniczonej liczby połączeń HTTP przy użyciu klienta HTTP Apache (wersja 3.x). Ponieważ wymyślenie dobrej konfiguracji zajęło mi kilka godzin, podzielę się:

private ExecutorService executor = new ThreadPoolExecutor(5, 10, 60L,
  TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
  Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

Tworzy to, ThreadPoolExecutorktóry zaczyna się od pięciu i przechowuje maksymalnie dziesięć jednocześnie uruchomionych wątków używanych CallerRunsPolicydo wykonywania.

Zgodnie z Javadoc dla ThreadPoolExecutor:

Jeśli jest więcej niż corePoolSize, ale mniej niż maximumPoolSize wątków, nowy wątek zostanie utworzony tylko wtedy, gdy kolejka jest pełna . Ustawiając corePoolSize i maximumPoolSize na takie same, tworzysz pulę wątków o stałym rozmiarze.

(Podkreśl moje.)

Odpowiedź jittera jest tym, czego chcesz, chociaż moja odpowiada na twoje inne pytanie. :)

jest jeszcze jedna opcja. Zamiast używać nowej kolejki SynchronousQueue, możesz również użyć dowolnej innej kolejki, ale musisz upewnić się, że jej rozmiar to 1, aby zmusić usługę executorservice do utworzenia nowego wątku.

Nie wygląda na to, żeby którakolwiek z odpowiedzi faktycznie odpowiadała na pytanie - w rzeczywistości nie widzę sposobu na zrobienie tego - nawet jeśli podklasujesz z PooledExecutorService, ponieważ wiele metod / właściwości jest prywatnych, np. Tworzenie addIfUnderMaximumPoolSize było chronione, możesz wykonaj następujące czynności:

class MyThreadPoolService extends ThreadPoolService {
    public void execute(Runnable run) {
        if (poolSize() == 0) {
            if (addIfUnderMaximumPoolSize(run) != null)
                return;
        }
        super.execute(run);
    }
}

Najbliżej mi było to - ale nawet to nie jest zbyt dobrym rozwiązaniem

new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) {
    public void execute(Runnable command) {
        if (getPoolSize() == 0 && getActiveCount() < getMaximumPoolSize()) {        
            super.setCorePoolSize(super.getCorePoolSize() + 1);
        }
        super.execute(command);
    }

    protected void afterExecute(Runnable r, Throwable t) {
         // nothing in the queue
         if (getQueue().isEmpty() && getPoolSize() > min) {
             setCorePoolSize(getCorePoolSize() - 1);
         }
    };
 };

ps nie testował powyższego

Oto inne rozwiązanie. Myślę, że to rozwiązanie zachowuje się tak, jak chcesz (choć nie jest z niego dumne):

final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    public boolean offer(Runnable o) {
        if (size() > 1)
            return false;
        return super.offer(o);
    };

    public boolean add(Runnable o) {
        if (super.offer(o))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
};

RejectedExecutionHandler handler = new RejectedExecutionHandler() {         
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        queue.add(r);
    }
};

dbThreadExecutor =
        new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, queue, handler);

To jest to, czego chcesz (przynajmniej tak mi się wydaje). Aby uzyskać wyjaśnienie, sprawdź odpowiedź Jonathana Feinberga

Executors.newFixedThreadPool(int n)

Tworzy pulę wątków, która ponownie wykorzystuje stałą liczbę wątków działających poza udostępnioną kolejką nieograniczoną. W dowolnym momencie co najwyżej nThreads będzie aktywnych zadań przetwarzania. Jeśli dodatkowe zadania zostaną przesłane, gdy wszystkie wątki będą aktywne, będą czekać w kolejce, aż wątek będzie dostępny. Jeśli jakikolwiek wątek zakończy się z powodu awarii podczas wykonywania przed zamknięciem, nowy wątek zajmie jego miejsce, jeśli będzie to konieczne do wykonania kolejnych zadań. Wątki w puli będą istniały do ​​momentu jawnego zamknięcia.

1. Możesz użyć ThreadPoolExecutorzgodnie z sugestią @sjlee

   Możesz dynamicznie kontrolować wielkość puli. Spójrz na to pytanie, aby uzyskać więcej informacji:

   Dynamiczna pula wątków

   LUB

2. Możesz użyć interfejsu API newWorkStealingPool , który został wprowadzony w java 8.

   public static ExecutorService newWorkStealingPool()

   Tworzy pulę wątków do kradzieży pracy przy użyciu wszystkich dostępnych procesorów jako docelowego poziomu równoległości.

Domyślnie poziom równoległości jest ustawiony na liczbę rdzeni procesora na serwerze. Jeśli masz serwer z 4 rdzeniami procesora, rozmiar puli wątków wynosiłby 4. Ten interfejs API zwraca ForkJoinPooltyp ExecutorService i umożliwia kradzież pracy bezczynnych wątków poprzez kradzież zadań z zajętych wątków w ForkJoinPool.

Problem został podsumowany następująco:

Chcę czegoś dokładnie takiego jak buforowana pula wątków (tworzy wątki na żądanie, a następnie zabija je po pewnym czasie), ale z ograniczeniem liczby wątków, które może utworzyć i możliwością kontynuowania kolejkowania dodatkowych zadań po osiągnięciu limit wątku.

Zanim wskażę rozwiązanie, wyjaśnię, dlaczego poniższe rozwiązania nie działają:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());

Nie spowoduje to kolejkowania zadań po osiągnięciu limitu 3, ponieważ SynchronousQueue z definicji nie może przechowywać żadnych elementów.

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

Nie spowoduje to utworzenia więcej niż jednego wątku, ponieważ ThreadPoolExecutor tworzy tylko wątki przekraczające corePoolSize, jeśli kolejka jest pełna. Ale LinkedBlockingQueue nigdy nie jest pełny.

ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<Runnable>());
executor.allowCoreThreadTimeOut(true);

Nie spowoduje to ponownego użycia wątków, dopóki nie zostanie osiągnięty corePoolSize, ponieważ ThreadPoolExecutor zwiększa liczbę wątków do momentu osiągnięcia corePoolSize, nawet jeśli istniejące wątki są bezczynne. Jeśli możesz żyć z tą wadą, to jest to najłatwiejsze rozwiązanie problemu. Jest to również rozwiązanie opisane w artykule „Współbieżność języka Java w praktyce” (przypis na str. 172).

Wydaje się, że jedynym kompletnym rozwiązaniem opisanego problemu jest przesłonięcie metody kolejki offeri napisanie a, RejectedExecutionHandlerjak wyjaśniono w odpowiedziach na to pytanie: Jak sprawić, aby ThreadPoolExecutor zwiększył wątki do maksimum przed kolejkowaniem?

Działa to dla Java8 + (i innych, na razie ..)

     Executor executor = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>()){{allowCoreThreadTimeOut(true);}};

gdzie 3 to limit liczby wątków, a 5 to limit czasu dla bezczynnych wątków.

Jeśli chcesz sprawdzić, czy to działa samodzielnie , oto kod do wykonania pracy:

public static void main(String[] args) throws InterruptedException {
    final int DESIRED_NUMBER_OF_THREADS=3; // limit of number of Threads for the task at a time
    final int DESIRED_THREAD_IDLE_DEATH_TIMEOUT=5; //any idle Thread ends if it remains idle for X seconds

    System.out.println( java.lang.Thread.activeCount() + " threads");
    Executor executor = new ThreadPoolExecutor(DESIRED_NUMBER_OF_THREADS, DESIRED_NUMBER_OF_THREADS, DESIRED_THREAD_IDLE_DEATH_TIMEOUT, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>()) {{allowCoreThreadTimeOut(true);}};

    System.out.println(java.lang.Thread.activeCount() + " threads");

    for (int i = 0; i < 5; i++) {
        final int fi = i;
        executor.execute(() -> waitsout("starting hard thread computation " + fi, "hard thread computation done " + fi,2000));
    }
    System.out.println("If this is UP, it works");

    while (true) {
        System.out.println(
                java.lang.Thread.activeCount() + " threads");
        Thread.sleep(700);
    }

}

static void waitsout(String pre, String post, int timeout) {
    try {
        System.out.println(pre);
        Thread.sleep(timeout);
        System.out.println(post);
    } catch (Exception e) {
    }
}

wyjście powyższego kodu dla mnie to

1 threads
1 threads
If this is UP, it works
starting hard thread computation 0
4 threads
starting hard thread computation 2
starting hard thread computation 1
4 threads
4 threads
hard thread computation done 2
hard thread computation done 0
hard thread computation done 1
starting hard thread computation 3
starting hard thread computation 4
4 threads
4 threads
4 threads
hard thread computation done 3
hard thread computation done 4
4 threads
4 threads
4 threads
4 threads
3 threads
3 threads
3 threads
1 threads
1 threads