nowa metoda FixedThreadPool

Metoda newFixedThreadPool klasy Executors utworzy dla nas executorService ze stałą liczbą wątków. W porównaniu do metody newSingleThreadExecutor określamy ile wątków chcemy widzieć w puli. Pod maską nazywa się

new ThreadPoolExecutor(nThreads, nThreads,
                                      	0L, TimeUnit.MILLISECONDS,
                                      	new LinkedBlockingQueue());

Jako parametry corePoolSize (ile wątków będzie gotowych (uruchomionych) w momencie uruchomienia usługi executora ) oraz maximumPoolSize (maksymalna liczba wątków, które może utworzyć usługa executora ) przekazywana jest ta sama liczba - liczba wątków przekazana do newFixedThreadPool( nWątki) . W ten sam sposób możemy przekazać parametry i własną implementację ThreadFactory .

Zobaczmy więc, dlaczego potrzebujemy takiej usługi ExecutorService .

ExecutorService ze stałą liczbą (n) wątków ma następującą logikę:

• Maksymalnie n wątków będzie aktywnych do przetwarzania zadań.
• Jeśli przesłanych zostanie więcej niż n zadań, będą one przechowywane w kolejce do momentu zwolnienia wątków.
• Jeśli jeden z wątków ulegnie awarii i zakończy się, zostanie utworzony nowy wątek, który zastąpi uszkodzony.
• Każdy wątek z puli jest aktywny do momentu zamknięcia puli.

Jako przykład wyobraźmy sobie kolejkę na lotnisku, gdzie wszyscy stoją w jednej kolejce i tuż przed kontrolą rozchodzą się o roboczą liczbę punktów. Jeśli w jednym z punktów wystąpi opóźnienie, kolejka będzie przechodzić tylko przez drugi, aż pierwszy będzie wolny, a jeśli w ogóle zepsuje się jeden z punktów, to w jego miejsce zostanie otwarty inny punkt i nastąpi kontrola. kontynuować przez dwa punkty.

Od razu należy zaznaczyć, że choć warunki są idealne, gdzie obiecane n wątków działa stabilnie i zawsze znajdzie się podmiana wątków, które zakończyły się błędem (co już jest niemożliwe do osiągnięcia w rzeczywistej eksploatacji lotniska ze względu do ograniczonych zasobów), system nadal ma kilka nieprzyjemnych cech, więc jak nie będzie więcej wątków w żadnych okolicznościach, nawet jeśli kolejka rośnie szybciej niż wątki będą przetwarzać zadania.

Proponuję zrozumieć w praktyce, jak działa ExecutorService ze stałą liczbą wątków. Stwórzmy klasę, która implementuje Runnable . Obiekty tej klasy będą naszymi zadaniami dla ExecutorService .

public class Task implements Runnable {
    int taskNumber;
 
    public Task(int taskNumber) {
        this.taskNumber = taskNumber;
    }
 
    @Override
    public void run() {
try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Processed user request #" + taskNumber + " on thread " + Thread.currentThread().getName());
    }
}
    

W metodzie run() blokujemy wątek na 2 sekundy, symulując obciążenie i wyświetlamy numer bieżącego zadania oraz nazwę wątku, na którym wykonujemy to zadanie.

ExecutorService executorService = Executors.newFixedThreadPool(3);
 
        for (int i = 0; i < 30; i++) {
            executorService.execute(new Task(i));
        }
        
        executorService.shutdown();
    

Na początek stworzymy usługę ExecutorService i wyślemy 30 zadań do wykonania.

W konsoli widzimy, jak zadania są wykonywane w różnych wątkach, gdy są zwalniane z poprzedniego zadania.

Teraz zwiększymy liczbę zadań do 100, a po wysłaniu 100 zadań wywołamy metodę awaitTermination(11, SECONDS) . W parametrach podajemy ilość oraz jednostkę czasu. Ta metoda zablokuje główny wątek na 11 sekund, po czym wywołamy shutdownNow() i wymusimy zamknięcie usługi ExecutorService bez czekania na zakończenie wszystkich zadań.

ExecutorService executorService = Executors.newFixedThreadPool(3);
 
        for (int i = 0; i < 100; i++) {
            executorService.execute(new Task(i));
        }
 
        executorService.awaitTermination(11, SECONDS);
 
        executorService.shutdownNow();
        System.out.println(executorService);
    

Na koniec wyprowadzimy dane o stanie executorService .

W konsoli widzimy:

Po tym następuje 3 InterruptedException s , które rzucają metody uśpienia z 3 aktywnych zadań.

Widzimy, że w momencie zakończenia wykonaliśmy 15 zadań, ale w puli nadal są 3 aktywne wątki, które nie zakończyły wykonywania swojego zadania. Te 3 wątki wywołują przerwanie() , co oznacza, że ​​zadanie zostanie zakończone, ale w naszym przypadku metoda sleep rzuca nam InterruptedException . Widzimy też, że po wywołaniu metody shutdownNow() kolejka zadań została wyczyszczona.

Warto więc zauważyć, że zasada działania musi być brana pod uwagę przy korzystaniu z usługi ExecutorService ze stałą liczbą wątków w puli. Warto używać tego typu do zadań o znanym stałym obciążeniu.

Jest jeszcze jedno ciekawe pytanie: jeśli potrzebujesz użyć executora dla 1 wątku, którą metodę wywołać - newSingleThreadExecutor() czy newFixedThreadPool(1) ?

W zachowaniu oba executory będą równoważne. Jedyną różnicą jest to, że metoda newSingleThreadExecutor() zwróci taki executor, którego w przyszłości nie będzie można przekonfigurować do obsługi dodatkowych wątków.