Урок 9 — Конкурентност: проверка на всичко наведнъж

Време: 2 × 90 мин (две сесии; разделянето е между раздели 9.4 и 9.5)

Какво изграждаш: конкурентният revisor: проверява всички услуги едновременно, с лимит, който ти задаваш, и предава отчета в същия ред както винаги

Какво научаваш: платформени и виртуални нишки; една виртуална нишка за задача и никога в пул (pool); ExecutorService като ресурс, който се затваря; invokeAll, стабилният ред и ExecutionException; Semaphore за ограничаване; състезание (race condition), предизвикано и измерено; ConcurrentHashMap и атомарните типове; ScopedValue; и структурираната конкурентност като „какво предстои“

Откъде идваш. Носиш проекта от Урок 8: revisor в Maven, с HttpClient, Jackson, конфигурацията, валидирана в servicios.json, и 25 теста в зелено. Прави истинска заявка за всяка услуга, мери колко е продължила и не се срива, когато някоя услуга не отговаря. Но ги прави една след друга: с четирите канонични услуги отнема колкото сборът на времената им, около 2.5 секунди, а с двайсет услуги по половин секунда всяка би отнело десет. Тази бавност не е на програмата, а на формата ѝ: for на revisarTodos чака отговора на catalogo, преди да попита inventario, макар нито една от двете да не се нуждае от другата. Днес ѝ я отнемаш. Работиш в папката си ~/curso-java/revisor/.

Какво не прави този урок. Не добавя опции на командния ред: лимитът на конкурентността е константа до Урок 10, където става --concurrencia. Не променя и начина, по който се проверява една услуга: revisar остава такъв, какъвто е, с таймаутите си и трите си причини за падане. Единственото, което се променя, е кой вика revisar и колко пъти едновременно.

За стартиранията. Фигури 9.1 до 9.8 са компактни файлове и се стартират с java figNN_MM.java от папката на урока; не се нуждаят от лабораторията. Фигура 9.9 е проектът и тя се нуждае: в друг терминал, от ~/curso-java/, остави да върви java Laboratorio.java (слуша на порт 8089; порт 8088 остава празен нарочно, той е на reportes). Всички изходи са получени с Temurin 25.0.4 на машина с 10 ядра. Времената, които виждаш в този урок, зависят от твоята машина: това, което се повтаря навсякъде, са пропорциите, не милисекундите. Числата, които се променят от едно стартиране на друго (като резултата от едно състезание), са представени като фрагмент, с истинско стартиране, а не като проверен изход.

След урока ще можеш да

• Обясниш какво е виртуална нишка, с какво се различава от платформена нишка и защо програма, която чака много (като revisor), се възползва от тях.
• Стартираш по една задача за всяка услуга с Executors.newVirtualThreadPerTaskExecutor() вътре в try-with-resources и кажеш защо не се ползва пул от виртуални нишки.
• Ползваш invokeAll, за да изчакаш всички задачи и да вземеш резултатите им в реда на входа, и прочетеш ExecutionException до причината ѝ.
• Ограничиш колко задачи напредват едновременно със Semaphore, като винаги го освобождаваш във finally.
• Предизвикаш състезание (race condition), измериш колко се губи и го поправиш с ConcurrentHashMap.merge или с атомарен брояч.
• Обясниш защо revisor няма състезания (не споделя променливо състояние) и за какво служат ScopedValue и структурираната конкурентност.
• Измериш с часовника ефекта на лимита върху двайсет услуги и обясниш защо лимит 10 отнема двойно повече от лимит 20.

Защо, преди как

Представи си кухня с един-единствен готвач, който приготвя четири ястия. Слага първото във фурната и остава да я гледа двайсет минути, докато е готово; едва тогава започва второто. Никой не би казал, че този готвач е бавен с ръцете. Бавен е, защото чака, без да прави нищо друго. revisor от урок 8 е този готвач: когато пита inventario, която отговаря за 2.3 секунди, програмата стои 2.3 секунди, с почти заспал процесор, докато pagos и catalogo биха могли да отговарят едновременно.

Решението също е от кухнята: да сложиш няколко ястия във фурната наведнъж. В програмирането се нарича конкурентност (concurrency): няколко задачи в ход едновременно, така че времето, което една прекарва в чакане, да се използва от друга. (Не е същото като паралелизма (parallelism), който е да имаш няколко процесора, които работят буквално едновременно. Конкурентността е начин за организиране на програмата; паралелизмът — нещо, което оборудването може или не може да предложи. В този урок ще видиш и двете, но ползата за revisor идва най-вече от първата.)

В Java това винаги е имало препятствие и този урок съществува заради начина, по който Java го премахна. Повече от двайсет години единицата на конкурентността беше нишката на операционната система: нишката е независима линия на изпълнение вътре в програмата и всяка нишка на Java беше нишка на системата, със собствен стек и свои ресурси на операционната система, и с реална цена. (Собствена купчина (heap) нямаше: обектите живеят там и всички нишки ги споделят, а точно оттам идват състезанията от 9.5.) Да създадеш десет хиляди от тях за десет хиляди заявки не беше осъществимо, така че стандартната практика беше да има няколко „постоянни“ нишки (пул) и задачите да се редуват, с опашки, лимити и много код, за да не остане никой да чака повече от нужното. От Java 21 платформата предлага нещо друго: виртуални нишки, толкова евтини, че всяка задача може да има своя, и с тях кодът отново е такъв, какъвто беше в кухнята с един готвач —ред след ред, всяка стъпка изчаква предишната—, само че сега готвачите са много и евтини.

Този урок има две сесии. Първата (9.1 до 9.4) учи виртуалните нишки, invokeAll и семафора и завършва с измерване: revisor проверява двайсет услуги наведнъж. Втората (9.5 до 9.8) учи какво трябва да се знае, за да не те предаде конкурентността: споделеното състояние, състезанието, което го прави опасно, стойностите с обхват и какво предстои в платформата. Това е урокът, който най-много променя това, което знаеш за Java, и този, който най-много ще ти послужи: всеки съвременен сървър на Java се пише с тези идеи.

Понятията

9.1 Два вида нишка

9.1.1 Какво е нишка и колко струва

Нишката (thread) е последователност от инструкции, която напредва по своя сметка. Програмата ти винаги има поне една: тази, която изпълнява main. С повече нишки няколко последователности напредват „едновременно“: операционната система разпределя времето на процесорите между тях и когато една се блокира —например чака отговор от мрежата—, дава реда на друга.

До Java 20 всеки Thread беше точно това: нишка на операционната система, със собствен стек от извиквания и собствен запис в таблиците на системата. Тези сега се наричат платформени нишки (platform threads). Мощни са и скъпи: системата налага таван на броя нишки, всяка резервира памет за стека си и преминаването от една към друга струва. Затова сървърна програма не можеше да си позволи нишка за заявка, когато заявките бяха хиляди, и техниката беше да се споделят с пул.

Виртуалната нишка е нещо друго, макар да се ползва по същия начин. Също е Thread —същият тип, същите методи, join, interrupt—, но я управлява виртуалната машина на Java вместо операционната система. Една виртуална нишка не е вързана за нишка на системата: JVM има малка група платформени нишки, наречени носещи нишки (carrier threads), и „монтира“ върху тях виртуалните нишки, които имат какво да правят. Когато една виртуална нишка стигне до блокираща операция —четене от сокет, заспиване, чакане на семафор—, JVM я демонтира: запазва състоянието ѝ в паметта на купа (heap), освобождава носещата нишка и монтира друга виртуална нишка, която може да напредва. Когато отговорът пристигне, виртуалната нишка отново се монтира върху някоя носеща. Това е същата идея като при готвача, който не стои и не гледа фурната, но направена от платформата, а не от теб. Официалната документация за виртуалните нишки я описва в тези термини, а спецификацията на промяната е JEP 444, окончателна от Java 21.

Фигура 9.1 ги поставя една до друга. Иска платформена нишка и виртуална със същия конструктор на нишки (Thread.ofPlatform() и Thread.ofVirtual()), дава им име, стартира ги и всяка казва какво е:

// fig09_01.java
void main() throws InterruptedException {
  Thread plataforma =
      Thread.ofPlatform()
          .name("de-plataforma")
          .start(() -> describir("de-plataforma"));
  plataforma.join();

  Thread virtual =
      Thread.ofVirtual()
          .name("virtual-1")
          .start(() -> describir("virtual-1"));
  virtual.join();
}

void describir(String nombre) {
  Thread hilo = Thread.currentThread();
  IO.println(nombre + ": virtual=" + hilo.isVirtual() + ", demonio=" + hilo.isDaemon());
}

Фигурите на този урок нямат import: един компактен файл импортира сам всичко, което експортира java.base, включително java.util.concurrent. Методът join() кара main да изчака нишката да свърши; без него програмата би могла да приключи, преди нишката да отпечата. Ето какво отпечатва:

$ java fig09_01.java
de-plataforma: virtual=false, demonio=false
virtual-1: virtual=true, demonio=true

Три данни в два реда. Платформената нишка е virtual=false; другата — virtual=true. И виртуалната е демон (demonio=true; demonio е испанската дума за „демон“): една нишка демон не поддържа жива JVM, така че ако main приключи и останат само виртуални нишки, програмата свършва, без да ги чака. Това е разлика, която хапе: ако стартираш виртуална нишка „да работи във фонов режим“ и main приключи, ти я прекъсваш. (Затова изпълнителят от 9.2 се затваря, като изчаква задачите си.)

Можеш да видиш механиката на носещите нишки във всяка виртуална нишка. Ако отпечаташ Thread.currentThread() от една, излиза нещо такова:

VirtualThread[#26]/runnable@ForkJoinPool-1-worker-1

След знака @ е носещата нишка, върху която е вървяла в този момент. Когато стартирах двайсет хиляди виртуални нишки и записах носещата им, на моята машина се появиха точно десет имена, от worker-1 до worker-10: толкова, колкото са ядрата. Двайсет хиляди „нишки на Java“, разпределени върху десет „нишки на системата“.

9.1.2 Какво се печели: мащаб, не бързина

Фигура 9.2 поставя разликата в числа, със задача, която не прави нищо друго освен да чака 200 милисекунди (услуга, която се бави да отговори). Стартира хиляда такива два пъти: веднъж в класически пул от 50 платформени нишки и веднъж с по една виртуална нишка за задача. Мери всяка със системния часовник (System.nanoTime()):

// fig09_02.java
static final int TAREAS = 1_000;
static final int ESPERA_MS = 200;

void esperar() {
  try {
    Thread.sleep(ESPERA_MS);
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
  }
}

long medirMs(ExecutorService ejecutor) {
  long inicio = System.nanoTime();
  try (ejecutor) {
    for (int i = 0; i < TAREAS; i++) {
      ejecutor.submit(this::esperar);
    }
  }
  return (System.nanoTime() - inicio) / 1_000_000;
}

void main() {
  long conPool = medirMs(Executors.newFixedThreadPool(50));
  long conVirtuales = medirMs(Executors.newVirtualThreadPerTaskExecutor());
  IO.println("pool de 50 hilos de plataforma, más de 3 s: " + (conPool > 3000));
  IO.println("un hilo virtual por tarea, menos de 1 s: " + (conVirtuales < 1000));
}

Няколко подробности за четенето, преди да я стартираш. try (ejecutor) { ... } ползва променлива, която вече съществува, като ресурс на try-with-resources (Java го позволява от 9, докато променливата не се присвоява отново); това е същата идея от урок 5 и при излизане от блока се вика close(). this::esperar е препратка към метод на този същия клас, която Java превръща в задача. А двата последни реда отпечатват булева стойност, а не милисекундите, за да е изходът винаги един и същ:

$ java fig09_02.java
pool de 50 hilos de plataforma, más de 3 s: true
un hilo virtual por tarea, menos de 1 s: true

Истинските милисекунди ги виждаш, ако отпечаташ променливите conPool и conVirtuales. В три поредни стартирания ми дадоха 4,152, 4,148 и 4,161 ms за пула и 221, 214 и 206 ms за виртуалните нишки. Аритметиката обяснява първото: хиляда задачи на 50 нишки са 20 партиди, а 20 партиди по 200 ms са 4 секунди. С една виртуална нишка за задача няма партиди: хилядата чакат заедно и общото време е това на едно чакане, плюс цената на стартирането. При друго измерване сто хиляди виртуални нишки, заспиващи по една секунда всяка, завършиха за 1.4 секунди.

Тук е добре да се спре едно недоразумение, което разваля повече от един дизайн. Документацията на Oracle го казва без заобикалки: виртуалните нишки не са по-бързи нишки; не изпълняват кода по-бързо от платформена. Дават мащаб (повече задачи в ход едновременно), а не бързина (по-малко време на задача). И служат за задачи, които прекарват почти цялото си време чакайки, а не за работа на процесора: ако всяка задача беше дълго разлагане на прости множители, десет хиляди виртуални нишки нямаше да свършат по-рано от десет, защото ядрата са същите. revisor е идеалният случай: всяка задача изпраща заявка и чака.

Още едно нещо, което ти спестява грижи, които ще четеш в по-стари книги: с Java 21 имаше проблем, наречен pinning (виртуална нишка, която блокираше вътре в блок synchronized, оставаше залепена за носещата си и не я освобождаваше). От Java 24 вече не се случва в тези блокове; съветът „избягвай synchronized с виртуални нишки“, който още циркулира, е от друга епоха.

9.2 Една виртуална нишка за задача

9.2.1 Изпълнителят, който затваря

Да създаваш нишки на ръка с Thread.ofVirtual().start(...) служи да разбереш, не да организираш работа. За да я организираш, се ползва изпълнител (executor), обект, на когото предаваш задачи и който се грижи да ги изпълни. Фабриката, която ни интересува, е един ред:

try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
  // aquí se mandan tareas con ejecutor.submit(...) o ejecutor.invokeAll(...)
}

Името ѝ го казва всичко: virtual thread per task executor, „изпълнител с една виртуална нишка за задача“. Всяка задача, която му изпратиш, получава нова виртуална нишка, която умира, когато задачата свърши. Няма опашка, няма размер на пула, няма нищо за конфигуриране. И тъй като ExecutorService е AutoCloseable от Java 19, try-with-resources прави важното: при излизане от блока вика close(), който чака всички задачи да свършат, преди да продължи. Това е гаранцията, която старите изпълнители нямаха, където забравянето на shutdown() оставяше живи нишки и програма, която не приключваше. Тук блокът try е ограда: това, което започнеш вътре, приключва вътре.

Правилото на този раздел се събира в едно изречение и е това, което програмата на курса поставя в получер шрифт: една виртуална нишка за задача, никога в пул. Един пул съществува, за да разпределя оскъден ресурс —платформените нишки— между много задачи. Виртуалните нишки не са оскъдни: могат да се имат милиони. Ако им сложиш пул отгоре, пак създаваш изкуствен недостиг, със същите опашки и същите тавани, и губиш точно това, което ги правеше полезни. Официалното ръководство го обобщава така: виртуалните нишки не бива да представляват споделен ресурс, а задача. Ако това, което искаш, е да ограничиш колко задачи напредват едновременно, има инструмент, направен за това, и го виждаш в 9.4.

9.2.2 Защо блокиращият send е наред

Погледни отново как revisor проверява една услуга от урок 8: cliente.send(peticion, BodyHandlers.discarding()) блокира: нишката, която го вика, остава да чака, докато дойде отговорът или изтече времето. В платформена нишка това беше скъп лукс; затова конкурентното програмиране в Java години наред търсеше начини да не блокира (callbacks, CompletableFuture, реактивни библиотеки), с код, който се чете наопаки. Във виртуална нишка чакането на вход или изход като този струва много малко: JVM демонтира нишката, докато чака, и ползва носещата за друга. Малко не е нищо: има случаи, в които виртуалната нишка остава фиксирана (pinned) към носещата си и я заема, докато чака (например по време на извикване на нативен код), а това, което има от другата страна, връзки, услуги, памет, си остава ограничено: евтината нишка не прави евтин сървъра, който питаш. Затова revisar не променя нито ред в този урок. Остава линейният метод от урок 8: сглобява заявката, изпраща я, чака, класифицира. Конкурентността не е вътре в revisar; тя е в начина, по който се вика.

Две неща от това извикване трябва да се помнят. Първо, HttpClient се споделя: документацията на класа казва, че един екземпляр е неизменяем и служи за изпращане на много заявки, така че revisor създава само един и всички задачи го ползват; да създадеш клиент за заявка би значело да изхвърлиш преизползваемите му връзки. Второ, клиентът също е AutoCloseable и редът на затваряне има значение: в try-with-resources с два ресурса те се затварят в обратен ред на декларирането. Затова клиентът се декларира първи, а изпълнителят после: при излизане се затваря изпълнителят (изчаква всички задачи) и после клиентът. Обратното би затворило клиента с заявки в полет.

9.3 invokeAll и стабилният ред

9.3.1 Callable, Future и изчакване на всички

submit приема два вида задача. Един Runnable не връща нищо и не може да хвърли проверявано изключение. Един Callable<T> връща стойност от тип T и може да хвърли всяко изключение: това е, от което се нуждае revisar, който връща Resultado и може да хвърли InterruptedException (урок 5 обясни защо то никога не се поглъща). Това, което получаваш, когато изпратиш задача, е Future<T>: обещание за резултат. futuro.get() изчаква —ако е нужно— задачата да свърши и връща стойността ѝ.

Методът, който съединява двете неща, е invokeAll: получава колекция от Callable, изпълнява всичките, изчаква всички да свършат и връща List<Future<T>>. Важното е в документацията и това го прави подходящ за revisor: списъкът, който връща, е в същия ред като този, който му дадеш, независимо коя е свършила първа. Фигура 9.3 го проверява с три задачи, които отнемат 300, 50 и 150 ms; най-бързата е втората в списъка и въпреки това излиза втора:

// fig09_03.java
String tarea(String nombre, long esperaMs) throws InterruptedException {
  Thread.sleep(esperaMs);
  return nombre;
}

void main() throws Exception {
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Callable<String>> tareas =
        List.of(
            () -> tarea("lenta", 300),
            () -> tarea("rapida", 50),
            () -> tarea("mediana", 150));

    for (Future<String> futuro : ejecutor.invokeAll(tareas)) {
      IO.println("en el orden de la lista: " + futuro.get());
    }

    List<Callable<String>> conFallo =
        List.of(
            () -> tarea("buena", 10),
            () -> {
              throw new IllegalStateException("fallé a propósito");
            });
    List<Future<String>> futuros = ejecutor.invokeAll(conFallo);
    IO.println("invokeAll regresó; la tarea 2 terminó: " + futuros.get(1).isDone());
    try {
      futuros.get(1).get();
    } catch (ExecutionException e) {
      IO.println("get() lanza " + e.getClass().getSimpleName());
      IO.println("la causa es " + e.getCause());
    }
  }
}

Следвай фигурата на части. tarea заспива и връща името си. Списъкът от три Callable се пише с ламбди, а invokeAll ги изпълнява всички с виртуални нишки. Първият for обхожда Future-ите по ред. После се сглобява втори списък, в който една задача хвърля изключение, и се вижда нещо важно: invokeAll се връща нормално, задачата се появява като завършена и провалът не избухва, докато не поискаш стойността ѝ с get(). Това е стартирането:

$ java fig09_03.java
en el orden de la lista: lenta
en el orden de la lista: rapida
en el orden de la lista: mediana
invokeAll regresó; la tarea 2 terminó: true
get() lanza ExecutionException
la causa es java.lang.IllegalStateException: fallé a propósito

Първите три реда излизат в реда на списъка (lenta, rapida, mediana), а не в реда, в който задачите са свършили (който беше rapida, mediana, lenta). За revisor това е гаранцията за точка 3 от „Как разбирам, че съм успял“: отчетът излиза винаги в същия ред, този на конфигурацията, дори inventario да отнема две секунди, а pagos осемдесет и седем милисекунди. Ако събираше резултатите, докато идват, редът би се променял при всяко стартиране и да сравниш два отчета би било мъка.

9.3.2 ExecutionException: изключението на друга нишка

Когато една задача хвърли изключение в своята нишка, това изключение не може да се издигне по стека на този, който я е изпратил —те са различни стекове—. Java го пази вътре във Future и при извикване на get() го връща обвито в ExecutionException. Последните два реда на фигурата го показват: типът, който улавяш, е ExecutionException, но важното е неговата причина, getCause(), която е оригиналното изключение (IllegalStateException: fallé a propósito, т.е. „провалих се нарочно“). Който го улови и отпечата само съобщението на обвивката, губи точно информацията, от която има нужда. Това е същият принцип от урок 5, този за собствените изключения с причината им: едно обвито изключение се разгръща, не се изхвърля.

Това те задължава да решиш какво да правиш с него и решението на revisor е обмислено. Има два вида проблем. Паднала услуга е очакван резултат: revisar я улавя и я превръща в Resultado със състояние CAIDO и причината си и никога не оставя да излезе IOException. Затова никоя задача не пропада по вина на мрежата и отчетът излиза пълен, дори половин свят да е паднал. Грешка на програмата (NullPointerException, логическа грешка) е друго нещо: ако задача пропадне така, нещо в кода не е наред и да го скриеш зад „паднала услуга“ би било лъжа. revisor го хвърля отново като IllegalStateException с приложена оригиналната причина. В едно изречение: паднала услуга не събаря отчета; грешка на програмата не се скрива.

Една последна подробност от договора: ако нишката, която чака в invokeAll, бъде прекъсната, методът отменя задачите, които не са свършили, и хвърля InterruptedException. Затова сигнатурата на revisarTodos декларира throws InterruptedException и не го поглъща.

9.4 Ограничаване със Semaphore

9.4.1 За какво да ограничаваш, ако нишките са евтини

Ако виртуалните нишки са почти безплатни, за какво да ограничаваш? Защото оскъдният ресурс вече не е нишката; той е това, което има от другата страна. Двайсет услуги, проверявани едновременно, е разумна учтивост. Двайсет хиляди услуги, проверявани едновременно —или двайсет едновременни заявки към същия сървър—, могат да съборят една крехка услуга, да изчерпят портовете на системата или да задействат алармата на доставчик. Лимитът защитава целта, не JVM. И това решение не бива да зависи от броя нишки, които случайно има един пул.

Java има инструмент, предвиден за това от двайсет години: Semaphore. Семафорът пази число разрешения (permits). Който иска да напредне, вика acquire() (взема разрешение); ако няма, чака, докато друг го освободи с release(). С new Semaphore(2) най-много две задачи са вътре в защитената зона; останалите стоят на опашка. Фигура 9.4 стартира шест задачи от по 100 ms със семафор с две разрешения и записва максимума от едновременни задачи с AtomicInteger (цяло число, което може да се променя от няколко нишки, без да се губят сметки; връща се в 9.5):

// fig09_04.java
final AtomicInteger activas = new AtomicInteger();
final AtomicInteger maximo = new AtomicInteger();

void trabajar() throws InterruptedException {
  int ahora = activas.incrementAndGet();
  maximo.accumulateAndGet(ahora, Math::max);
  Thread.sleep(100);
  activas.decrementAndGet();
}

void main() throws Exception {
  var semaforo = new Semaphore(2);
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 6; i++) {
      ejecutor.submit(
          () -> {
            semaforo.acquire();
            try {
              trabajar();
            } finally {
              semaforo.release();
            }
            return null;
          });
    }
  }
  IO.println("tareas lanzadas: 6, máximo al mismo tiempo: " + maximo.get());
}

Забележи формата на задачата: acquire(), после try с работата и release() във finally. Не е въпрос на стил, а условие за коректност. Ако работата хвърли изключение и release() не е във finally, това разрешение се губи завинаги; след достатъчно провали всички разрешения са изгубени и останалите задачи чакат вечно, програма, която е увиснала без никаква видима грешка. Стартирането:

$ java fig09_04.java
tareas lanzadas: 6, máximo al mismo tiempo: 2

Шест стартирани задачи, никога повече от две едновременно. И забележи какво не направи семафорът: не ограничи нишките. Шестте задачи имат виртуалната си нишка от самото начало; четири са заспали в acquire() и не струват нищо. Официалното ръководство на Oracle препоръчва именно това —семафор за ограничаване— и обяснява, че ефектът му прилича на този на пул с фиксиран размер, защото и в двата случая излишните задачи стоят на опашка, но без да резервират нишки за това.

9.4.2 Защо не пул от виртуални нишки

Изкушението е да решиш лимита с това, което вече познаваш: Executors.newFixedThreadPool(2, fabrica), където фабриката произвежда виртуални нишки. Работи и това е капанът. Фигура 9.5 мери максимума от едновременни задачи по двата начина:

// fig09_05.java
final AtomicInteger activas = new AtomicInteger();
final AtomicInteger maximo = new AtomicInteger();

void trabajar() {
  int ahora = activas.incrementAndGet();
  maximo.accumulateAndGet(ahora, Math::max);
  try {
    Thread.sleep(100);
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
  }
  activas.decrementAndGet();
}

int maximoCon(ExecutorService ejecutor) {
  maximo.set(0);
  try (ejecutor) {
    for (int i = 0; i < 6; i++) {
      ejecutor.submit(this::trabajar);
    }
  }
  return maximo.get();
}

void main() {
  var fabrica = Thread.ofVirtual().factory();
  int conPool = maximoCon(Executors.newFixedThreadPool(2, fabrica));
  int sinPool = maximoCon(Executors.newVirtualThreadPerTaskExecutor());
  IO.println("pool fijo de 2 hilos virtuales: " + conPool);
  IO.println("un hilo virtual por tarea: " + sinPool);
}

$ java fig09_05.java
pool fijo de 2 hilos virtuales: 2
un hilo virtual por tarea: 6

Фиксираният пул ограничава до 2, изпълнителят за задача пропуска всичките 6. Има два проблема с ползването на първия за ограничаване. Единият е за яснота: лимитът е скрит в числов аргумент на една фабрика и никой, който чете кода, няма да разбере, че „2“ означава „позволени едновременни заявки“. Другият е инженерен: току-що си въвел отново опашка, размер за настройване и нишка за резервиране, за да получиш това, което Semaphore(2) дава с един ред и с целта си, изписана в името. Официалното ръководство е категорично: виртуалните нишки никога не бива да се групират. По-малко конкурентност, повече код, никакво предимство.

9.4.3 Стъпката на проекта: конкурентният revisarTodos

Вече имаш частите. Така остава Revision.java; новото са константата LIMITE_POR_OMISION и методът revisarTodos, а всичко останало (revisar, nuevoCliente) е от урок 8. Чети revisarTodos по ред: първо валидира лимита; създава семафора; отваря клиента и изпълнителя (в този ред, както видя в 9.2.2); сглобява по една задача за услуга, която взема разрешение, проверява и го освобождава във finally; изпраща всичките с invokeAll; и събира резултатите по ред, разгръщайки всяка ExecutionException:

// fig09_09/src/main/java/mx/habil/revisor/Revision.java

package mx.habil.revisor;

import java.io.IOException;
import java.net.ConnectException;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.net.http.HttpTimeoutException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import mx.habil.revisor.Respuesta.Respondio;
import mx.habil.revisor.Respuesta.SinRespuesta;

/** Consulta los servicios por HTTP y arma sus resultados. */
final class Revision {

  /** Cuánto se espera para establecer la conexión, antes de decir que no hay respuesta. */
  static final long TIEMPO_CONEXION_MS = 2000;

  /** Cuántas peticiones puede haber a la vez si nadie dice otra cosa. */
  static final int LIMITE_POR_OMISION = 10;

  private Revision() {}

  /** Un cliente HTTP con tiempo de conexión y sin seguir redirecciones. */
  static HttpClient nuevoCliente() {
    return HttpClient.newBuilder()
        .connectTimeout(Duration.ofMillis(TIEMPO_CONEXION_MS))
        .followRedirects(HttpClient.Redirect.NEVER)
        .build();
  }

  /** Consulta un servicio y mide cuánto tarda; ninguna falla de red sale de este método. */
  static Resultado revisar(HttpClient cliente, Servicio servicio) throws InterruptedException {
    HttpRequest peticion =
        HttpRequest.newBuilder(servicio.url())
            .timeout(Duration.ofMillis(servicio.timeoutMs()))
            .GET()
            .build();
    long inicio = System.nanoTime();
    Respuesta respuesta;
    try {
      HttpResponse<Void> http = cliente.send(peticion, BodyHandlers.discarding());
      long tiempoMs = (System.nanoTime() - inicio) / 1_000_000;
      respuesta = new Respondio(http.statusCode(), tiempoMs);
    } catch (HttpTimeoutException e) {
      respuesta = new SinRespuesta("tiempo agotado");
    } catch (ConnectException e) {
      respuesta = new SinRespuesta("sin conexión");
    } catch (IOException e) {
      respuesta = new SinRespuesta("error de red");
    }
    return new Resultado(servicio, Clasificador.clasificar(respuesta), respuesta);
  }

  /**
   * Revisa todos los servicios a la vez, con a lo más {@code limite} peticiones en vuelo.
   *
   * @param servicios los servicios, en el orden en que debe salir el reporte
   * @param limite cuántas peticiones puede haber al mismo tiempo; al menos 1
   * @return los resultados, en el mismo orden que {@code servicios}
   * @throws InterruptedException si alguien interrumpe al programa mientras espera
   */
  static List<Resultado> revisarTodos(List<Servicio> servicios, int limite)
      throws InterruptedException {
    if (limite < 1) {
      throw new IllegalArgumentException("el límite debe ser al menos 1");
    }
    var semaforo = new Semaphore(limite);
    try (var cliente = nuevoCliente();
        var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
      var tareas = new ArrayList<Callable<Resultado>>();
      for (Servicio servicio : servicios) {
        tareas.add(
            () -> {
              semaforo.acquire();
              try {
                return revisar(cliente, servicio);
              } finally {
                semaforo.release();
              }
            });
      }
      var resultados = new ArrayList<Resultado>();
      for (Future<Resultado> futuro : ejecutor.invokeAll(tareas)) {
        try {
          resultados.add(futuro.get());
        } catch (ExecutionException e) {
          throw new IllegalStateException("la revisión de un servicio falló", e.getCause());
        }
      }
      return resultados;
    }
  }
}

Три решения на тази функция заслужават да се погледнат отблизо. Първото: семафорът се взема вътре в задачата, не преди да се стартира. Всички задачи съществуват от началото и всяка чака реда си сама; invokeAll остава прост. Второто: времето, което revisar мери със System.nanoTime(), започва след вземането на разрешението, така че колоната ВРЕМЕ (TIEMPO) продължава да казва колко е отнела услугата, а не колко е чакала на опашка. Третото: limite < 1 се отхвърля още на входа с IllegalArgumentException, защото семафор с нула разрешения би оставил всички задачи да чакат завинаги.

Константата LIMITE_POR_OMISION е 10 и Revisor я ползва засега. Revisor се променя в две неща: първият аргумент, ако го има, е конфигурационният файл (така можеш да проверяваш друг списък, без да редактираш нищо), и вика revisarTodos с лимита по подразбиране:

// fig09_09/src/main/java/mx/habil/revisor/Revisor.java

package mx.habil.revisor;

import java.nio.file.Path;
import java.util.List;

/** El revisor: lee servicios.json, consulta cada servicio y escribe la tabla. */
public final class Revisor {

  private Revisor() {}

  /**
   * Punto de entrada del programa.
   *
   * @param args el primero, si lo hay, es el archivo de configuración
   * @throws InterruptedException si alguien interrumpe al programa mientras espera
   */
  public static void main(String[] args) throws InterruptedException {
    List<Servicio> servicios;
    try {
      Path ruta = args.length > 0 ? Path.of(args[0]) : Path.of("servicios.json");
      servicios = Configuracion.leer(ruta, Configuracion.SERVIDORES_PERMITIDOS);
    } catch (ErrorDeConfiguracion e) {
      System.err.println("error de configuración: " + e.getMessage());
      return;
    }
    var resultados = Revision.revisarTodos(servicios, Revision.LIMITE_POR_OMISION);
    System.out.println(Formato.tabla(Reporte.de(resultados)));
  }
}

Файлът servicios-20.json има двайсет услуги, servicio01 до servicio20, всичките насочени към маршрута /espera/500 на лабораторията, който отговаря 200 след половин секунда. Така е цял:

{
  "servicios": [
    {"nombre": "servicio01", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio02", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio03", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio04", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio05", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio06", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio07", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio08", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio09", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio10", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio11", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio12", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio13", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio14", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio15", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio16", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio17", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio18", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio19", "url": "http://127.0.0.1:8089/espera/500"},
    {"nombre": "servicio20", "url": "http://127.0.0.1:8089/espera/500"}
  ]
}

Измерване: часовникът решава

С включена лаборатория компилирай (mvn -B verify) и стартирай revisor срещу тези двайсет услуги, като мериш колко отнема цялата програма. В Linux, в терминал с bash, чистият начин за измерване е запазената дума time с кратък формат:

$ TIMEFORMAT='%R s'
$ time java -cp "target/classes:target/lib/*" mx.habil.revisor.Revisor servicios-20.json > /dev/null
1.258 s

Промени константата LIMITE_POR_OMISION в Revision.java (засега няма друг начин; в Урок 10 ще е --concurrencia), прекомпилирай и повтори. Това са моите измервания, две стартирания за лимит, винаги с двайсетте услуги по 500 ms:

Лимит	Партиди	Какво предсказва аритметиката	Какво измерих (и двете стартирания)
1 (равносилно на последователно)	20	20 × 0.5 = 10 s	10.413 s и 10.387 s
3	7	7 × 0.5 = 3.5 s	3.808 s и 3.799 s
10	2	2 × 0.5 = 1 s	1.258 s и 1.298 s
20	1	1 × 0.5 = 0.5 s	0.751 s и 0.779 s

Две неща, за да четеш таблицата, без да се заблудиш. Първо: измереното винаги надхвърля предсказаното с около 0.2 секунди и това е фиксираната цена за стартиране на JVM и на HTTP клиента. Проверих го: една-единствена услуга с нулево чакане отнема 0.2 секунди общо. Извади 0.2 от всяко измерване и аритметиката се изпълнява с разлика от една-две десети. Второ: последната колона отговаря на въпроса от програмата на курса, „защо лимит 10 отнема двойно повече от лимит 20“. С лимит 10 и двайсет услуги има две партиди: първите десет вземат разрешение, чакат половин секунда; когато го освободят, влизат другите десет и чакат още половин секунда. С лимит 20 влизат всички наведнъж. Общото време е броят партиди по продължителността на всяко чакане, а не броят услуги.

С четирите канонични услуги, обратно, подобрението е скромно: минават от около 2.8 секунди последователно на около 2.5. Не е дефект: времето вече не е сборът (2.5 s), а това на най-бавната (inventario, 2.3 s), и тъй като тя командва, има малко за печелене. Конкурентността помага, когато има много сходни чакания; с една доминираща почти не помага. Важно е да го знаеш, защото „направи се конкурентно“ не означава „стана бързо“: мери се.

И с четирите канонични услуги на servicios.json, от папката на проекта и с включена лаборатория, едно пълно стартиране изглежда така:

$ cd fig09_09
$ java -cp "target/classes:target/lib/*" mx.habil.revisor.Revisor
SERVICIO    ESTADO    TIEMPO  DETALLE
catalogo    ARRIBA     176ms  200
inventario  LENTO       2.3s  200
pagos       ARRIBA     122ms  200
reportes    CAIDO          —  sin conexión

4 servicios · 2 arriba · 1 lento · 1 caído · promedio 878ms

Времената в колоната ВРЕМЕ (TIEMPO) са от моето стартиране; твоите ще са различни, но СЪСТОЯНИЕТО (ESTADO) и ПОДРОБНОСТТА (DETALLE) не се променят никога.

(Тук свършва първата сесия.)

9.5 Споделено състояние: където конкурентността хапе

9.5.1 Състезание, предизвикано и измерено

Всичко дотук беше лесно, защото задачите не се докосват помежду си. Щом две нишки пишат едни и същи данни, се появява проблемът, който дава лоша слава на конкурентността. Вижда се най-добре, като го предизвикаш. Тази програма стартира 10,000 задачи и всяка прибавя 100 пъти единица към брояч, който споделят. Ако всичко мине добре, общото би било точно един милион. Запази я като Carrera.java (carrera е „състезание“):

int contador = 0;

void main() {
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
      ejecutor.submit(
          () -> {
            for (int j = 0; j < 100; j++) {
              contador++;
            }
          });
    }
  }
  IO.println("esperado 1000000, contado " + contador);
}

Нарочно няма маркер за фигура: резултатът ѝ се променя при всяко стартиране, така че не може да се провери срещу фиксиран изход. Стартирах програмата четири пъти поред и това преброи (esperado е „очаквано“, contado — „преброено“):

esperado 1000000, contado 841360
esperado 1000000, contado 852985
esperado 1000000, contado 773685
esperado 1000000, contado 855739

Няма нито едно изключение, няма предупреждение на компилатора и програмата „работи“: приключва, отпечатва число, излиза с код 0. Но изгуби между 144,000 и 226,000 прибавяния и не губи едни и същи два пъти. Това е чертата, която прави едно състезание опасно: не се обявява. Програма със състезание минава тестовете си в 99 % от случаите и пропада в продукция, във вторник, под натоварване.

Защо се случва, е по-просто, отколкото изглежда. contador++ изглежда като една операция, но са три: да прочетеш стойността, да ѝ прибавиш единица и да запишеш резултата. Ако две нишки прочетат 41 едновременно, и двете изчисляват 42 и двете записват 42: прибавено е веднъж вместо два пъти и едно прибавяне е изчезнало. С десет носещи нишки, работещи наистина паралелно, това застъпване се случва стотици хиляди пъти. (Има и втора причина, тази на видимостта: без инструкция, която го урежда, паметта на една нишка може да не види това, което е записала друга. Моделът на паметта на Java, в глава 17 на спецификацията на езика, определя кога един запис е гарантирано видим: правилото се нарича happens-before.)

Не е проблем само на числата. С една споделена колекция е по-лошо. Същият експеримент с HashMap —записва 10,000 различни ключа, всеки от своята задача— би трябвало да завърши с 10,000 записа. Запази го като MapaCompartido.java:

void main() {
  var mapa = new HashMap<Integer, Integer>();
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
      int clave = i;
      ejecutor.submit(() -> mapa.put(clave, clave));
    }
  }
  IO.println("esperadas 10000, hay " + mapa.size());
}

Трите ми стартирания завършиха с 8,339, 7,195 и 7,563 записа. Липсваха между 1,600 и 2,800 записа. Един HashMap не е проектиран да се пише от няколко нишки и вътрешната му структура се разваля: освен това може да хвърли изключения и, в по-стари версии на Java, да остане в безкраен цикъл. Това, че понякога „излиза добре“, не е утеха; това е, което го прави трудно за залавяне.

9.5.2 Три начина да се поправи, по ред на предпочитание

Първият и най-добрият: да не споделяш. Ако всяка задача работи със своите данни и връща резултат, няма какво да се защитава. Това е, което прави revisor, и го обяснявам в 9.5.3.

Вторият: типове, направени да се споделят. Пакетът java.util.concurrent носи типове, чиито операции са атомарни, тоест неделими. AtomicInteger и AtomicLong са цели числа, които могат да се събират от няколко нишки, без да се губят сметки; LongAdder, предвиден за много оспорвани броячи, е по-бърз, когато има много нишки. А ConcurrentHashMap е картата за конкурентни записи. Методът му merge е този, който поправя типичния случай на броене по категория: mapa.merge("ARRIBA", 1, Integer::sum) означава „ако ключът не съществува, сложи му 1; ако съществува, замени го с резултата от събирането“; и според документацията цялото извикване се изпълнява атомарно. Фигура 9.6 повтаря експеримента с броенето на състояния с 10,000 задачи, сега с merge, и събира и с атомарен брояч:

// fig09_06.java
void main() {
  var porEstado = new ConcurrentHashMap<String, Integer>();
  var total = new AtomicInteger();
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
      ejecutor.submit(
          () -> {
            porEstado.merge("ARRIBA", 1, Integer::sum);
            total.incrementAndGet();
          });
    }
  }
  IO.println(porEstado);
  IO.println("total atómico: " + total.get());
}

$ java fig09_06.java
{ARRIBA=10000}
total atómico: 10000

За разлика от състезанието, този резултат излиза винаги един и същ, колкото пъти го стартираш: 10,000 в картата и 10,000 в брояча. Това вече може да се провери и затова тази фигура има маркер.

Третият: ключалки (locks). Java има synchronized и ключалките на java.util.concurrent.locks, за да защитят парче код така, че само една нишка наведнъж да го изпълнява. Работят и понякога са единственото, което върши работа; но са инструментът, който най-лесно се ползва зле (взаимни блокировки, забравени ключалки, твърде дълги секции), и курсът не ги ползва. С двете предишни опции почти никога не са нужни.

9.5.3 Защо revisor няма състезания

Прегледай всяко парче данни, което задачите на revisarTodos докосват, едно по едно, защото тук се плаща това, което научи в урок 3. Servicio, който получава всяка задача, е record: неизменяем, никой не го променя. HttpClient е безопасен за споделяне (документацията го гарантира). Задачата не пише в никоя външна променлива: връща своя Resultado, който е друг неизменяем record. Единственият променлив обект, който задачите споделят, е Semaphore, който е направен точно за това. А списъкът resultados, който е обикновен ArrayList (не е безопасен за няколко нишки), се докосва само от нишката, която е извикала revisarTodos, след като invokeAll е приключил; самият пакет java.util.concurrent документира, че това, което една задача е направила преди да свърши, е гарантирано видимо за този, който взема резултата ѝ с get().

Правилото за дизайн, което излиза оттук, струва повече от всеки инструмент: комуникирай, като връщаш стойности, не като споделяш памет. Когато една задача трябва да каже нещо на този, който я е стартирал, нека го върне. Неизменяемостта, за която курсът се грижи от урок 3, не беше стилово украшение: тя е това, което позволява едно парче да се изпълнява в сто нишки, без да мислиш за тях.

9.6 Стойности с обхват: ScopedValue

Има проблем, който конкурентните задачи изостря: контекстът. Представи си, че всяка проверка носи идентификатор на изпълнението, r-001, и искаш всеки метод —този, който класифицира, този, който сглобява заявката, този, който пише диагностично съобщение— да може да го прочете. Очевидният изход е да се подава като параметър през всеки метод от веригата и това се превръща в шум, който никой не иска. Обичайният изход в Java беше ThreadLocal, променлива, която има различна стойност във всяка нишка. Но има три известни недостатъка: променлива е (всеки я променя и никой не знае кой), живее докато някой я почисти (забравянето е изтичане на памет) и копирането ѝ към дъщерни нишки е скъпо, което тежи, когато са милиони виртуални нишки.

За това съществува ScopedValue, окончателна от Java 25 (JEP 506). Една стойност с обхват е неизменяема и съществува само докато трае един блок код: свързваш я с ScopedValue.where(CLAVE, valor).run(...) (или .call(...), ако блокът връща нещо или хвърля изключения) и всеки метод, извикан отвътре, може да я прочете с CLAVE.get(). При излизане от блока стойността изчезва сама. Фигура 9.7 я показва и показва и нейната граница:

// fig09_07.java
static final ScopedValue<String> CORRIDA = ScopedValue.newInstance();

void anotar(String mensaje) {
  IO.println("[corrida " + CORRIDA.get() + "] " + mensaje);
}

void main() throws Exception {
  ScopedValue.where(CORRIDA, "r-001")
      .call(
          () -> {
            anotar("revisando 4 servicios");
            Thread otro =
                Thread.ofVirtual()
                    .start(() -> IO.println("en un hilo nuevo, ligado = " + CORRIDA.isBound()));
            otro.join();
            return null;
          });
  IO.println("fuera del bloque, ligado = " + CORRIDA.isBound());
}

$ java fig09_07.java
[corrida r-001] revisando 4 servicios
en un hilo nuevo, ligado = false
fuera del bloque, ligado = false

Методът anotar не получава идентификатора като параметър и го чете въпреки това: това е услугата, която предоставя. Вторият ред е частта, която изненадва: виртуалната нишка, създадена вътре в блока на ръка, не вижда стойността (ligado = false; ligado е „свързано“). Стойността пътува към извикванията на една и съща нишка, не към нишките, които стартираш по своя инициатива; само структурираната конкурентност от 9.7 я наследява на децата си. А третият ред потвърждава, че извън блока не остава нищо.

revisor на този курс не ползва ScopedValue: няма никакъв контекст за предаване. Ще го видиш веднага щом напишеш сървър, който обслужва заявки на няколко потребители и трябва да знаеш, във всяка точка на кода, за кого работиш. Тук е, защото е това, което замества ThreadLocal в Java, която ще пишеш оттук нататък, и защото дебютира в тази версия.

9.7 Какво предстои: структурираната конкурентност

В 9.3 видя, че invokeAll чака всички задачи, каквото и да стане. Липсва една форма, която много хора искат: „стартирай тези задачи като единица: ако една пропадне, отмени останалите; ако блокът свърши, нито една да не остава жива“. Това се нарича структурирана конкурентност, и JDK 25 я носи... в предварителен преглед (preview). Един предварителен преглед е завършена функция, но все още подлежаща на промени, която се включва само с флаг. Тази е сменяла формата си във всяка версия от 21; петият предварителен преглед е JEP 505.

Фигура 9.8 ползва API (StructuredTaskScope.open(), с fork за стартиране и join за чакане). Компилирана така, не върви:

// fig09_08.java
String consultar(String nombre) {
  return "respuesta de " + nombre;
}

void main() throws Exception {
  try (var alcance = StructuredTaskScope.open()) {
    var catalogo = alcance.fork(() -> consultar("catalogo"));
    var pagos = alcance.fork(() -> consultar("pagos"));
    alcance.join();
    IO.println(catalogo.get() + " / " + pagos.get());
  }
}

$ java fig09_08.java
fig09_08.java:7: error: StructuredTaskScope is a preview API and is disabled by default.
  try (var alcance = StructuredTaskScope.open()) {
                     ^
  (use --enable-preview to enable preview APIs)
1 error
error: compilation failed

Да прочетеш съобщението ти казва всичко: класът съществува, но е API в предварителен преглед и е изключен. С флага, който самото съобщение иска, работи:

$ java --enable-preview --source 25 fig09_08.java
respuesta de catalogo / respuesta de pagos

Не я ползвай в revisor, нито в продукция: API може да се промени между версиите на JDK и програма, която зависи от него, би трябвало да се пренапише, когато се промени. Затова този курс я оставя като любопитство. Което взимаш със себе си, е идеята: когато се стабилизира, ще е естественият начин да кажеш „тези три запитвания са една работа“, а invokeAll със Semaphore ще е бил твоят начин да стигнеш дотам. Дотогава това, което научи, работи в Java 21, 25 и следващите.

9.8 Тестовете на конкурентния revisor

Конкурентният revisor носи три нови теста, в класа RevisionConcurrenteTest. Всеки тества едно от обещанията на 9.3 и 9.4 и нито един не зависи от часовника повече от необходимото. Ето какво проверяват: редът (най-бавната услуга, поставена първа в списъка, остава първа в резултата); лимитът (с лимит 2 и осем заявки, сървърът на теста никога не вижда повече от две едновременно, брои ги един AtomicInteger вътре в самия сървър); и валидацията (лимит 0 е IllegalArgumentException). Забележи, че сървърът на теста се създава с Executors.newVirtualThreadPerTaskExecutor(): сървърът на com.sun.net.httpserver, ако не му се даде изпълнител, обслужва заявките една по една в една нишка и тестът за лимита би измерил 1 вместо 2, без семафорът да има нищо общо.

// fig09_09/src/test/java/mx/habil/revisor/RevisionConcurrenteTest.java

package mx.habil.revisor;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

class RevisionConcurrenteTest {

  private static HttpServer servidor;
  private static int puerto;
  private static final AtomicInteger ACTIVAS = new AtomicInteger();
  private static final AtomicInteger MAXIMO = new AtomicInteger();

  @BeforeAll
  static void levantarServidor() throws IOException {
    servidor = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
    servidor.createContext("/lento", i -> responder(i, 400));
    servidor.createContext("/rapido", i -> responder(i, 0));
    servidor.createContext("/contado", RevisionConcurrenteTest::responderContando);
    servidor.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
    servidor.start();
    puerto = servidor.getAddress().getPort();
  }

  @AfterAll
  static void apagarServidor() {
    servidor.stop(0);
  }

  private static void responder(HttpExchange intercambio, long esperaMs) throws IOException {
    try {
      Thread.sleep(esperaMs);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
    intercambio.sendResponseHeaders(200, -1);
    intercambio.close();
  }

  private static void responderContando(HttpExchange intercambio) throws IOException {
    int ahora = ACTIVAS.incrementAndGet();
    MAXIMO.accumulateAndGet(ahora, Math::max);
    try {
      responder(intercambio, 200);
    } finally {
      ACTIVAS.decrementAndGet();
    }
  }

  private static Servicio servicio(String nombre, String ruta) {
    return new Servicio(nombre, URI.create("http://127.0.0.1:" + puerto + ruta));
  }

  @Test
  void elMasLentoPrimeroEnLaListaSigueSiendoElPrimeroEnElResultado() throws Exception {
    var servicios = List.of(servicio("lento", "/lento"), servicio("rapido", "/rapido"));
    List<Resultado> resultados = Revision.revisarTodos(servicios, 10);
    assertEquals("lento", resultados.get(0).servicio().nombre());
    assertEquals("rapido", resultados.get(1).servicio().nombre());
    assertEquals(Estado.ARRIBA, resultados.get(1).estado());
  }

  @Test
  void conLimiteDosElServidorNuncaVeMasDeDosPeticionesALaVez() throws Exception {
    var servicios = new ArrayList<Servicio>();
    for (int i = 1; i <= 8; i++) {
      servicios.add(servicio("s" + i, "/contado"));
    }
    MAXIMO.set(0);
    Revision.revisarTodos(servicios, 2);
    assertEquals(2, MAXIMO.get());
  }

  @Test
  void unLimiteMenorQueUnoSeRechaza() {
    assertThrows(
        IllegalArgumentException.class,
        () -> Revision.revisarTodos(List.of(servicio("s", "/rapido")), 0));
  }
}

Тест, който никога не си виждал да пропада, не доказва нищо, затова направи експеримента. В Revision.java смени new Semaphore(limite) с new Semaphore(Integer.MAX_VALUE) (семафор с толкова разрешения, че никога не спира) и стартирай тестовете. Това казва отчетът на surefire:

[ERROR] Tests run: 3, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 0.737 s <<< FAILURE! -- in mx.habil.revisor.RevisionConcurrenteTest
[ERROR] mx.habil.revisor.RevisionConcurrenteTest.conLimiteDosElServidorNuncaVeMasDeDosPeticionesALaVez -- Time elapsed: 0.220 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <2> but was: <8>

Червен стана точно тестът за лимита и с числото, което трябваше: без спирачка, осемте заявки стигнаха едновременно. Върни семафора към стойността му и отчетът се връща към зелено. За да преброиш тестовете, без да вярваш на кода на изход (exit code) на Maven, събери ги от отчета, по два различни начина (броене, което решава нещо, се прави два пъти):

$ grep -h -o ' tests="[0-9]*"' target/surefire-reports/TEST-*.xml | awk -F'"' '{s+=$2} END {print s}'
28
$ cat target/surefire-reports/TEST-*.xml | grep -c '<testcase '
28

Двайсет и пет от урок 8 плюс трите от днес: 28 и нито един пропуснат.

pom.xml се променя само в едно: версията става 0.9.0. Извън файловете, които видя цели в този урок (Revision.java, Revisor.java, RevisionConcurrenteTest.java и servicios-20.json), целият проект остава точно както беше в урок 8, без да пипаш ред.

Грешката, която ще видиш

Грешката на този урок не е съобщение, а липса на съобщение: състезанието от 9.5.1. Програмата завършва без изключения и с грешно число (esperado 1000000, contado 773685, т.е. „очаквано 1000000, преброено 773685“). Ако в някой момент твоя програма даде различни резултати при еднакви стартирания, подозирай състезание преди всичко друго и търси кои данни пишат две нишки едновременно. Четири спомагателни съобщения, за да ги разпознаваш с един поглед:

• java.util.concurrent.ExecutionException: java.lang.IllegalStateException: fallé a propósito: една задача е пропаднала в друга нишка. Причината е след двоеточията; виж getCause() (fig09_03.java).
• AssertionFailedError: expected: <2> but was: <8>: тестът за лимита ти казва, че семафорът не е спрял (9.8).
• Споделен HashMap, който завършва с по-малко записи от тези, които си записал (esperadas 10000, hay 7195, т.е. „очаквани 10000, има 7195“): картата се е развалила; ползвай ConcurrentHashMap.
• error: StructuredTaskScope is a preview API and is disabled by default.: ползвал си функция в предварителен преглед без флага (fig09_08.java).

Какво се прави погрешно

Поставяне на виртуалните нишки в пул. newFixedThreadPool(10, Thread.ofVirtual().factory()) ограничава, но скрито, с опашка и размер, които някой ще трябва да настройва. Една виртуална нишка за задача и лимитът със Semaphore (9.4.2).

Освобождаване на разрешението извън finally. Изключение между acquire() и release() губи разрешението. След няколко провала всички задачи чакат завинаги и програмата увисва без обяснение (9.4.1).

Поглъщане на InterruptedException. Един празен catch (InterruptedException e) {} скрива сигнала, че някой иска задачата да спре. Ако не можеш да го разпространиш, възстанови флага с Thread.currentThread().interrupt() (урок 5).

Споделяне на колекция, която не е конкурентна. ArrayList, HashMap и HashSet не са направени да се пишат от няколко нишки. Ползвай тези от java.util.concurrent или, по-добре, не споделяй (9.5).

Ползване на виртуални нишки за работа на процесора. Не дават бързина, дават мащаб. Една задача с дълго изчисление заема носещата си през цялото време и не отстъпва реда; за това важи броят на ядрата (9.1.2).

Стартиране без лимит срещу чужда услуга. Това, че нишките са евтини, не значи, че целта издържа. Лимитът се поставя, като се мисли за услугата, която получава заявките, а не за JVM (9.4.1).

Пазене на контекст в променлив ThreadLocal и забравяне да се почисти. За това, което трае един блок, ScopedValue (9.6).

Тестване на конкурентност с Thread.sleep, за да се „даде време“. Тест, който чака фиксирано време и приема, че вече се е случило това, което чака, се чупи на бавна машина. Чакай това, което наистина трябва да се чака (join, invokeAll, close), както правят тестовете от 9.8.

Упражнения

Упражнение 1 — Измери трите времена

С включена лаборатория и servicios-20.json измери revisor с LIMITE_POR_OMISION на 1, на 10 и на 20. Преди да стартираш, запиши предсказанието си за всяко и обясни със свои думи защо лимит 10 отнема двойно повече от лимит 20. Запиши трите истински времена в дневника си, на реда на урок 9.

Упражнение 2 — Предскажи 3

Без още да го стартираш: колко трябва да отнеме revisor с лимит 3 върху същите двайсет услуги? Запиши сметката (колко партиди, колко трае всяка). После го измери, извади цената за стартиране на JVM и сравни. Ако предсказанието ти е сбъркало с повече от няколко десети, потърси какво си пропуснал.

Упражнение 3 — Поправи двете състезания

Вземи Carrera.java и MapaCompartido.java от 9.5.1 и ги поправи, всяко с различен инструмент от 9.5.2. Трябва да постигнеш брояч, който дава точно 1,000,000, и карта, която има точно 10,000 записа, при всяко стартиране. Стартирай своята версия три пъти, за да видиш, че не се променя.

Упражнение 4 — Разрешение, което се губи, и тестът, който го хваща

Трите теста от 9.8 не откриват лошо поставен release(). Напиши четвърти, който го открива: две услуги, лимит 1 и първата с адрес, който кара revisar да пропадне с изключение, което не е мрежово (например ftp://127.0.0.1/x, което HttpRequest отхвърля при сглобяването на заявката). Преди да го напишеш, предскажи какво би станало, ако semaforo.release() беше извън finally. После го извади от finally, виж теста в червено и го върни.

Решения

Решение 1

Моите времена бяха 10.4 s, 1.27 s и 0.76 s (виж таблицата от 9.4.3). Правилните предсказания бяха 10, 1 и 0.5 секунди плюс около 0.2 за стартиране. С лимит 10 има две партиди по половин секунда, с лимит 20 — една; времето е партиди × чакане, не услуги × чакане. Ако твоето число за лимит 20 е много по-голямо от 0.7 s, виж дали лабораторията не е обслужвала нещо друго.

Решение 2

Двайсет услуги на три разрешения са седем партиди (шест пълни и една от две): 7 × 0.5 = 3.5 s, плюс около 0.2 s за стартиране, тоест 3.7 s. Измерих 3.808 s и 3.799 s. Излишната десета от секундата са разходи за операции (отваряне на връзки, разпределяне на нишки), които не измерих поотделно; важното е, че сметката за партидите излезе вярна. Обърни внимание на последната партида: макар да има само две услуги, тя чака цялата половин секунда, защото една партида трае колкото най-бавната си услуга, а не колкото всеки от членовете си.

Решение 3

За брояча — LongAdder (върши работа и AtomicLong); за картата — ConcurrentHashMap. Двете промени се събират в една програма:

void main() {
  var contador = new LongAdder();
  var mapa = new ConcurrentHashMap<Integer, Integer>();
  try (var ejecutor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
      int clave = i;
      ejecutor.submit(
          () -> {
            for (int j = 0; j < 100; j++) {
              contador.increment();
            }
            mapa.put(clave, clave);
          });
    }
  }
  IO.println("esperado 1000000, contado " + contador.sum());
  IO.println("esperadas 10000 entradas, hay " + mapa.size());
}

Трите стартирания дадоха:

esperado 1000000, contado 1000000
esperadas 10000 entradas, hay 10000

Урокът е, че поправката не е „да внимаваш повече“, а да смениш типа на споделените данни с такъв, направен да се споделя.

Решение 4

Предсказанието: с лимит 1 първата услуга влиза, revisar хвърля изключението и тъй като release() е останал извън finally, никога не се изпълнява: единственото разрешение е изгубено. Втората услуга остава да чака в acquire() завинаги, invokeAll никога не се връща и програмата увисва. Тестът, с таван на времето от JUnit, за да се превърне увисването в провал със съобщение:

  @Test
  @Timeout(5)
  void unErrorDelProgramaNoDejaPermisosPerdidos() {
    var servicios =
        List.of(
            new Servicio("malo", URI.create("ftp://127.0.0.1/x")), servicio("rapido", "/rapido"));
    assertThrows(IllegalStateException.class, () -> Revision.revisarTodos(servicios, 1));
  }

(Нужен е import org.junit.jupiter.api.Timeout;.) С finally на мястото си тестът минава: първата услуга пропада и освобождава разрешението си, втората се проверява и revisarTodos хвърля IllegalStateException, което вече познаваш. С release() извън finally отчетът казва:

[ERROR] mx.habil.revisor.RevisionConcurrenteTest.unErrorDelProgramaNoDejaPermisosPerdidos -- Time elapsed: 5.013 s <<< ERROR!
java.util.concurrent.TimeoutException: unErrorDelProgramaNoDejaPermisosPerdidos() timed out after 5 seconds

Без @Timeout същият този тест не би пропаднал: би увиснал, а увисване в непрекъснатата интеграция струва много повече от провал. Когато тестваш конкурентност, сложи таван на времето на всичко, което би могло да чака завинаги. Върни finally и тестът се връща към зелено (с четирите теста на този клас общото би било 29; фигурата на урока остава с трите оригинални).

Как разбирам, че съм успял

• java fig09_01.java отпечатва virtual=true, demonio=true за виртуалната нишка и можеш да обясниш защо виртуална нишка, която не се чака, се прекъсва, когато main приключи.
• java fig09_02.java отпечатва true в двата си реда и можеш да пресметнеш на ръка защо пул от 50 отнема четири секунди с хиляда задачи по 200 ms.
• В fig09_03.java трите резултата излизат в реда на списъка, а не в този на приключване, и можеш да кажеш каква е причината на една ExecutionException.
• java fig09_04.java съобщава, че максимумът едновременно е 2, и знаеш защо release() е във finally.
• Стартира Carrera.java няколко пъти и видя, че числото се променя и никога не стига 1,000,000.
• java fig09_06.java отпечатва {ARRIBA=10000} при всяко стартиране.
• mvn -B verify завършва в зелено и сборът от тестовете на отчета, преброен по два начина, е 28.
• С включена лаборатория servicios-20.json отнема около 10 s с лимит 1, близо до 1 s с лимит 10 и близо до 0.5 s с лимит 20 (плюс цената за стартиране на JVM) и ги записа в дневника.
• Със семафор без спирачка тестът за лимита става червен с expected: <2> but was: <8>.

Резюме

Отговори, без да гледаш урока:

1. Какво е носеща нишка и какво прави JVM, когато виртуална нишка стигне до блокираща операция?
2. Каква е разликата между „мащаб“ и „бързина“? За какъв вид задача служат виртуалните нишки и за какъв не?
3. Защо една виртуална нишка за задача, а не пул? Какво се ползва за ограничаване на колко задачи напредват едновременно?
4. В какъв ред методът invokeAll връща Future-ите и защо това има значение за отчета?
5. Какво съдържа една ExecutionException и защо трябва да се гледа причината ѝ? Какво решава revisor да прави с нея и защо?
6. Защо release() на един семафор е във finally? Какво става, ако не е?
7. Защо contador++ губи прибавяния, когато го изпълняват няколко нишки? Кои три начина да го поправиш познаваш и кой е предпочитаният?
8. Защо revisor няма състезания, макар да изпълнява десет задачи едновременно?
9. Какво има ScopedValue, което ThreadLocal няма, и защо една нишка, създадена на ръка вътре в блока, не вижда стойността?
10. Защо структурираната конкурентност остава „какво предстои“ в този курс?

За сравнение с други езици. Ако идваш от Go, задачата с една виртуална нишка за услуга е равностойна на стартирането на една goroutine за услуга и чакането с sync.WaitGroup; разликата е, че в Java резултатите се връщат в Future и по ред, а не през канал. Ако идваш от TypeScript, Promise.all е естественото сравнение, с разлика, която има значение: Promise.all отхвърля, щом едно обещание пропадне, докато invokeAll чака винаги всички. А в Python asyncio.TaskGroup отменя останалите задачи, ако една пропадне, което е точно това, което обещава структурираната конкурентност от 9.7. Курсовете по Go, TypeScript и Python решават същия този revisor със своите инструменти. А ако те интересува кога е удобно да се проверява на линия, както прави revisor, и кога е удобна опашка или събитие, статията стилове на интеграция: на линия, опашка или събитие го обсъжда с цените му.

За допълнително четене

• Oracle, „Virtual Threads“ (ръководство на JDK 25): какво са, за какво служат, защо не се групират и защо един семафор ограничава по-добре от един пул. На английски. Посетено на 7 октомври 2026 г.
• JEP 444, „Virtual Threads“ и JEP 491, „Synchronize Virtual Threads without Pinning“: спецификацията на виртуалните нишки и поправката на pinning в Java 24. Посетено на 7 октомври 2026 г.
• ExecutorService и ConcurrentHashMap в API на Java 25: точният договор на invokeAll, close и merge. Посетено на 7 октомври 2026 г.
• JEP 506, „Scoped Values“ и JEP 505, „Structured Concurrency (Fifth Preview)“: стойностите с обхват (окончателни в 25) и структурираната конкурентност (още в предварителен преглед). Посетено на 7 октомври 2026 г.