Пул от нишки и задачи

Код на:JavaC#PythonJavaScript

Накратко: Пулът от нишки е няколко работника пред опашка от задачи, вместо нова нишка за всяка задача; в Java това е ExecutorService, а резултатът идва като Future.

Проблем

Нишка за всяка снимка

Магазинът пуска разпродажба. Продавачите качват снимки на продуктите, а за всяка трябва умалено копие за списъка с продукти. Смаляването е работа, която смята: минава през всеки пиксел (виж работа, която чака, и работа, която смята).

Първият вариант пуска нишка за всяка качена снимка:

List<Thread> threads = new ArrayList<>();
for (String image : uploads) {
    Thread thread = new Thread(() -> saveThumbnail(image));
    thread.start();
    threads.add(thread);
}

С 20 снимки при теб всичко минава: „Готови: 20 умалени снимки“. В първия час на разпродажбата идват 10 000 наведнъж.

• Всяка нишка иска свой стек (обикновено до 1 MB) и място в операционната система. Около няколко хиляди нишки JVM-ът спира с OutOfMemoryError: unable to create native thread.
• Дори да има памет, 10 000 нишки на 8 ядра само се редуват. Процесорът губи време да ги превключва, а всяка снимка става по-бавна.

Сървърът не забавя, а пада. И падналият сървър не обслужва и купувачите.

Идеята

Няколко работника и опашка

Пул от нишки (thread pool) е няколко работника, създадени веднъж, и опашка от задачи пред тях. Задачата е „смали тази снимка“. Свободният работник взима следващата задача от опашката, свършва я и се връща за нова.

• Броят работници е постоянен. 10 000 снимки не правят 10 000 нишки, а 10 000 задачи в опашката. Задачата е малък обект, нишката е скъпа.
• Работниците се ползват отново. Нишката не се създава и не умира за всяка снимка.
• Опашката поема пика. Когато качванията са повече, отколкото работниците смогват, задачите чакат, вместо да сриват сървъра.

Това е производител и потребител: кодът, който приема качванията, слага задачи в опашката, а работниците са потребителите.

С пула вече се срещна в урока за нишките: ExecutorService, submit и Future.get(). Тук ще видим какво става вътре, как да подредиш резултатите, какво става с грешките и колко работника трябват.

Виж го

Трима работника

Смаляването тук е пауза, 2 ms за всеки KB, а размерът идва от името на снимката. Така изходът е един и същ при всяко пускане:

// Смаляване на снимка: тук е пауза, а размерът идва от името.
static int thumbnail(String image) throws InterruptedException {
    if (image.startsWith("повредена")) {
        throw new IllegalArgumentException("повреден файл");
    }
    int kb = 20 + 3 * image.length();
    Thread.sleep(2 * kb);
    return kb;
}

Осем снимки, пул от трима работника:

try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Общо: " + total + " KB");
}

Стъпка по стъпка, с думи

1. 8 снимки чакат. Пулът е 3 нишки: Executors.newFixedThreadPool(3).
2. submit прави всяка снимка задача в опашката. Трите нишки взимат първите три.
3. Нишка 3 свършва първа: яке е третата снимка и резултатът отива на третото място. Работникът не спира: взима следващата задача.
4. Опашката е празна. Нишка 1 свърши и няма какво да вземе, затова чака. Другите довършват последните снимки.
5. get() взима размерите по реда на снимките, не по реда на свършване. Общо 379 KB; краят на try затваря пула.

List<String> images = List.of("чанта.jpg", "обувки.jpg", "яке.jpg", "шапка.jpg", "часовник.jpg", "раница.jpg", "колан.jpg", "шал.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Общо: " + total + " KB");
}

Изход

чанта.jpg: 47 KB
обувки.jpg: 50 KB
яке.jpg: 41 KB
шапка.jpg: 47 KB
часовник.jpg: 56 KB
раница.jpg: 50 KB
колан.jpg: 47 KB
шал.jpg: 41 KB
Общо: 379 KB

• Executors.newFixedThreadPool(3) прави три нишки и опашка без край пред тях.
• submit слага задачата в опашката и веднага дава Future. Цикълът не чака нищо.
• get() чака точно своята задача. Цикълът ги чака по реда на снимките, затова и изходът е по него.
• try (var pool = …) затваря пула накрая: close() не приема нови задачи, изчаква започнатите и спира нишките. Това е от Java 19. В по-стар код ще видиш shutdown() и awaitTermination във finally.

По-точно, за любопитните

Опашката на newFixedThreadPool е без край. Ако качванията винаги идват по-бързо, отколкото пулът смогва, опашката расте, докато паметта свърши. Пул с ограничена опашка е new ThreadPoolExecutor(3, 3, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy()): при пълна опашка задачата се изпълнява в нишката, която я праща, и така приемането на качвания само се забавя.

Как е в другите езици: в C# басейнът е един и е общ (ThreadPool, Task.Run, Parallel), в Python е ThreadPoolExecutor, а Node няма готов пул от Worker-и. В проект там се ползва пакетът Piscina.

Пробвай

Намери грешката

Колега е пренаписал смаляването с пул, а разпродажбата не стана по-бърза. Шест снимки отнемат колкото без пул:

long start = System.nanoTime();
List<Integer> sizes = new ArrayList<>();
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        Future<Integer> size = pool.submit(() -> thumbnail(image));
        sizes.add(size.get());
    }
}
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("Отне над 0,5 s: " + (ms > 500));

Изход

Отне над 0,5 s: true

Избери реда с грешката.

1long start = System.nanoTime();2List<Integer> sizes = new ArrayList<>();3try (var pool = Executors.newFixedThreadPool(3)) {4 for (String image : images) {5 Future<Integer> size = pool.submit(() -> thumbnail(image));6 sizes.add(size.get());7 }8}9long ms = (System.nanoTime() - start) / 1_000_000;10System.out.println("Отне над 0,5 s: " + (ms > 500));

Провери

Идеята

Резултатите по реда на входа

Задачите свършват в ред, който не знаеш: зависи от снимките, от натоварването, от това кой работник е бил свободен. Затова никога не печатай „по реда на свършване“, ако искаш един и същ изход.

Два начина дават еднакъв резултат при всяко пускане:

• По реда на входа. Пазиш обещанията в списък по реда на задачите и ги чакаш в този ред. Така работи кодът отгоре.
• Сбор или брой. Ако ти трябва само общото, редът няма значение: 47 + 50 е колкото 50 + 47.

Понякога искаш резултатите по реда на свършване: за лента „готови 3 от 8“, или за да покажеш първите готови снимки. ExecutorCompletionService ги дава така. Тогава събираш само неща, които не зависят от реда:

try (var pool = Executors.newFixedThreadPool(3)) {
    var done = new ExecutorCompletionService<Integer>(pool);
    for (String image : images) {
        done.submit(() -> thumbnail(image));
    }
    int count = 0;
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        total += done.take().get();
        count++;
    }
    System.out.println("Готови: " + count + " от " + images.size() + ", общо: " + total + " KB");
}

Изход

Готови: 8 от 8, общо: 379 KB

ExecutorCompletionService пази своя опашка от готовите задачи. take() чака следващата готова, която и да е тя.

Идеята

Грешка в задачата

Една от снимките е повредена и thumbnail хвърля изключение. То не излиза в нишката, която е пуснала задачата: тя отдавна е продължила. Пулът го хваща и го пази в обещанието за резултата. Излиза, когато чакаш този резултат:

List<String> images = List.of("чанта.jpg", "обувки.jpg", "повредена.jpg", "яке.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    for (int i = 0; i < images.size(); i++) {
        try {
            int kb = sizes.get(i).get();
            System.out.println(images.get(i) + ": " + kb + " KB");
        } catch (ExecutionException e) {
            String why = e.getCause().getMessage();
            System.out.println(images.get(i) + ": грешка: " + why);
        }
    }
}

Изход

чанта.jpg: 47 KB
обувки.jpg: 50 KB
повредена.jpg: грешка: повреден файл
яке.jpg: 41 KB

get() хвърля ExecutionException, а истинското изключение е в getCause(). Другите снимки не страдат: пулът продължава със следващите задачи.

Пробвай

Къде отиде грешката

Втората снимка е повредена. Какво ще отпечата програмата?

List<String> images = List.of("чанта.jpg", "повредена.jpg", "яке.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        pool.submit(() -> thumbnail(image));
    }
}
System.out.println("Готово: " + images.size() + " снимки");

Отговор

Готово: 2 снимки

Готово: 3 снимки

Exception in thread "pool-1-thread-2" java.lang.IllegalArgumentException: повреден файл
Готово: 3 снимки

Провери

Идеята

Колко работника

Зависи от това какво правят:

• Работа, която смята (смаляване, PDF, компресия): колкото са ядрата. Девети работник на 8 ядра няма къде да върви, само се редува с другите.
• Работа, която чака (имейл, плащане, база): повече. Докато един работник чака мрежата, ядрото е свободно за друг. Ако задачата чака W ms и смята C ms, на ядро се събират около 1 + W / C работника.

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + waitMs / computeMs);
}

System.out.println("Снимки (само смятане): " + poolSize(8, 0, 50));
System.out.println("Имейли (90 ms чакане, 10 ms смятане): " + poolSize(8, 90, 10));

Изход

Снимки (само смятане): 8
Имейли (90 ms чакане, 10 ms смятане): 80

Броят ядра дава Runtime.getRuntime().availableProcessors(). Формулата е начало, не закон: измери. Ако имейлите ползват един сървър с ограничение от 20 връзки, 80 работника само ще чакат на опашка пред него.

Пробвай

Попълни

Колко работника за задача, която чака waitMs и смята computeMs?

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + );
}

Провери

По-точно, за любопитните

Виртуални нишки (Java 21) сменят сметката за работа, която чака. Виртуалната нишка струва няколкостотин байта, а когато чака (sleep, мрежа, база), JVM-ът я сваля от истинската нишка и пуска друга. Executors.newVirtualThreadPerTaskExecutor() дава нова виртуална нишка за всяка задача, и 10 000 имейла наистина са 10 000 нишки, без пул. За работа, която смята, те не помагат: ядрата са пак 8. А ресурсите зад тях (връзките към базата) пак се ограничават, например със Semaphore.

Упражнение

Фактурите на разпродажбата

Реши задачата в редактора си.

След разпродажбата всяка поръчка иска фактура. invoice(amount) прави PDF-а (тук е пауза от 50 ms) и връща сумата с 20% ДДС. Поръчка със сума 0 или по-малко е грешна и функцията хвърля изключение с текст „невалидна сума“.

Поръчките са 1001 до 1008 със суми 50, 125, 80, 0, 240, 35, 90 и 15 €. Направи фактурите в пул от 4 работника. Отпечатай ги по реда на поръчките, а грешната с текста на грешката. Накрая: сборът на добрите фактури и колко са с грешка.

Изход

Поръчка 1001: 60 €
Поръчка 1002: 150 €
Поръчка 1003: 96 €
Поръчка 1004: грешка: невалидна сума
Поръчка 1005: 288 €
Поръчка 1006: 42 €
Поръчка 1007: 108 €
Поръчка 1008: 18 €
Общо: 762 €, с грешка: 1

Още задачи

• Ограничена опашка. Направи пула така, че в опашката да чакат най-много 10 задачи. Какво става с единадесетата? (ArrayBlockingQueue и CallerRunsPolicy)
• Напредък. Отпечатвай „готови 3 от 8“ след всяка свършена фактура, а накрая пак подредения списък.
• Таймаут. Фактура, която се прави над 1 секунда, да се брои за грешна, без да спира останалите.

Тест

Провери се

1. Защо пул от няколко нишки, а не нишка за всяка качена снимка?

Нишка за всяка снимка дава грешни размериПулът прави всяка снимка по-бързаНишките са малко и се ползват отново, а опашката поема пика

2. Пул от 3 работника смалява 8 снимки. Третата свършва първа. Къде е размерът ѝ, когато ги взимаш по реда на списъка с get, result, Promise.all или sizes[i]?

На първото място: първа свършиНа третото място: редът е този на снимкитеЗависи от пускането

3. Задача в пула хвърля изключение. Кога го виждаш?

НикогаКогато чакаш резултата ѝ: get(), result(), awaitВеднага, програмата спира

4. Сървърът има 8 ядра. Всеки имейл чака мрежата 90 ms и смята 10 ms. Около колко работника трябват, за да са заети ядрата?

Около 80Около 10 000Точно 8

5. Колко работника за смаляване на снимки, ако то само смята, а сървърът има 8 ядра?

Около 80Около 8Колкото снимки има в опашката

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: Нишки · Защо нишки · Race condition · Производител и потребител
• Продължава в: async не е нишка · Как се тества конкурентен код · Когато съседът е бавен: timeout, retry, circuit breaker
• Задачи по темата: Нишки и race conditions