Как лучше работает ForkJoinPool.ManagedBlocker?
По вводным: насколько я понял, ForkJoinPool.ManagedBlocker - это отдельная блокирующая задача. С помощью реализации этого интерфейса ForkJoinPool определяет нужно ли создавать дополнительный поток для поддержания требуемого уровня параллелизма.
По вводным: но после выполнения следующего кода возникли вопросы:
public class MainClass { public static void main (String[] args) throws Exception { var pool = new ForkJoinPool(10); ForkJoinPool.managedBlock(new MyBlocker(1)); System.out.println("AAA"); } static class MyBlocker implements ForkJoinPool.ManagedBlocker { private int n; MyBlocker(int n) { this.n = n; } public boolean block() throws InterruptedException { System.out.println("MyBlocker block:" + Thread.currentThread().getName()); return false; } public boolean isReleasable() { System.out.println("MyBlocker isReleasable:" + Thread.currentThread().getName()); if(this.n==1) { try { Thread.sleep(2000); this.n--; return false; } catch(InterruptedException ex) { System.out.println(ex.getMessage()); } } return true; } } }Выведет:
По вводным: myBlocker isReleasable:main
MyBlocker block:main
По вводным: myBlocker isReleasable:main
AAA
Вопросы:
Нужно понять: 1. Почему задача выполняется не в потоке из пула, а в главном потоке и как заставить выполняться в пуле?
По вводным: 2. Если есть возможность выполнять потоки в пуле, как заставить выполняться в моем пуле? ForkJoinPool.managedBlock - статический метод.
Сейчас ситуация такая: 3. Есть ли возможность заставить выполняться задачу в пуле асинхронно? (не блокируя основной поток)
Опишите проблему, и специалист поможет с настройкой, исправлением ошибки или доработкой сайта. Подберём понятный план работ без лишней переписки.
Пока нет других ответов. Будьте первым, кто поможет автору.
Ответить на вопрос
ForkJoinPool.ManagedBlocker нужен не для “создания отдельной блокирующей задачи”, а для корректного сообщения ForkJoinPool: сейчас поток, который обслуживает work-stealing пул, может надолго заблокироваться, и пулу, возможно, нужно временно компенсировать это дополнительным worker-потоком. Это важно, когда внутри ForkJoinTask выполняется блокирующая операция: ожидание I/O, lock, queue.take(), Future.get() и т.п.
Ключевой момент в вашем примере: вы вызываете ForkJoinPool.managedBlock(new MyBlocker(1)) из main-потока, а не из worker-потока созданного pool. Поэтому созданный вами new ForkJoinPool(10) вообще не участвует в этой операции. managedBlock работает относительно текущего потока и текущего ForkJoinPool worker-контекста. Если вызов сделан из обычного main, ему нечего компенсировать в вашем pool.
Правильнее проверять так: отправить задачу в ForkJoinPool и уже внутри нее вызвать managedBlock.
ForkJoinPool pool = new ForkJoinPool(2); pool.submit(() -> { try { ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() { private boolean done; @Override public boolean block() throws InterruptedException { Thread.sleep(5000); done = true; return true; } @Override public boolean isReleasable() { return done; } }); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).join();
Метод isReleasable() должен быстро отвечать, можно ли уже не блокироваться. Метод block() выполняет реальное ожидание и возвращает true, когда блокировка закончилась. ForkJoinPool может вызывать isReleasable() больше одного раза, поэтому там не должно быть тяжелой логики.
ManagedBlocker не гарантирует, что новый поток будет создан в каждом случае. Он только дает пулу возможность принять такое решение. Если блокировка короткая или пул не считает компенсацию нужной, визуально вы можете не увидеть новых потоков. Для диагностики смотрите pool.getPoolSize(), getActiveThreadCount(), getQueuedTaskCount(), но учитывайте, что эти значения динамические.
Вывод: используйте ManagedBlocker только внутри задач ForkJoinPool, когда реально блокируете worker. Для обычного ExecutorService, CompletableFuture на другом пуле или main-потока этот механизм обычно не нужен и не даст ожидаемого эффекта.