Как ограничить concurrency при использовании участников в Scala?

Я прихожу из Java, где я отправлял Runnable в ExecutorService, поддерживаемый пулом потоков. В Java очень ясно, как установить ограничения на размер пула потоков.

Мне интересно использовать актеров Scala, но я не понимаю, как ограничить concurrency.

Скажем, гипотетически, что я создаю веб-сервис, который принимает "задания". Задание отправляется с запросами POST, и я хочу, чтобы моя служба запустила задание, а затем немедленно вернула 202 Accepted - то есть задания обрабатываются асинхронно.

Если я использую участников для обработки заданий в очереди, как я могу ограничить количество обрабатываемых одновременных заданий?

Я могу придумать несколько разных способов приблизиться к этому; Мне интересно, есть ли лучшая практика сообщества или, по крайней мере, некоторые четко установленные подходы, которые являются несколько стандартными в мире Scala.

Один из подходов, о котором я думал, состоит в том, что у него есть один координатор-актер, который будет управлять очередью заданий и участниками обработки работы; Я полагаю, что он мог бы использовать простое поле int для отслеживания того, сколько заданий обрабатывается в настоящее время. Я уверен, что в этом случае есть некоторые проблемы с этим подходом, например, чтобы отслеживать, когда возникает ошибка, чтобы уменьшить число. Вот почему мне интересно, если Scala уже предоставляет более простой или более инкапсулированный подход к этому.

BTW Я пытался задать этот вопрос некоторое время назад, но я спросил его плохо.

Спасибо!

Ответ 1

Вы можете переопределить системные свойства actors.maxPoolSize и actors.corePoolSize, которые ограничивают размер пула потоков актеров, а затем бросают столько же заданий в пул, сколько могут обрабатывать ваши актеры. Почему, по-вашему, вам нужно задушить ваши реакции?

Ответ 2

Я действительно рекомендую вам взглянуть на Akka, альтернативную реализацию Actor для Scala.

http://www.akkasource.org

У Akka уже есть интеграция JAX-RS [1], и вы можете использовать это совместно с LoadBalancer [2], чтобы уменьшить количество действий в параллели:

[1] http://doc.akkasource.org/rest [2] http://github.com/jboner/akka/blob/master/akka-patterns/src/main/scala/Patterns.scala

Ответ 3

У вас действительно есть две проблемы.

Во-первых, держит пул потоков, используемый субъектами под контролем. Это можно сделать, установив системное свойство players.maxPoolSize.

Второй - это безудержный рост числа задач, которые были отправлены в пул. Вы можете или не можете беспокоиться об этом, однако вполне возможно вызвать такие проблемы, как ошибки в памяти, а в некоторых случаях потенциально более тонкие проблемы, слишком быстро выполняя слишком много задач.

Каждый рабочий поток поддерживает удаление заданий. Dequeue реализуется как массив, который рабочий поток будет динамически увеличиваться до некоторого максимального размера. В 2.7.x очередь может расти сама по себе, и я видел, что триггер из-за ошибок памяти в сочетании с множеством параллельных потоков. Максимальный размер dequeue меньше 2.8. Dequeue также может заполнить.

Для решения этой проблемы необходимо контролировать, сколько задач вы создаете, что, вероятно, означает какой-то координатор, как вы описали. Я столкнулся с этой проблемой, когда актеры, которые инициируют какой-то конвейер обработки данных, намного быстрее, чем те, которые находятся позже. Для того, чтобы контролировать процесс, у меня обычно есть актеры позже в цепочке, отсылающие актеры ранее в цепочке каждые X-сообщения, и у них есть те, что были раньше в цепочке, останавливаются после X-сообщений и ждут ответа на пинг. Вы также можете сделать это с помощью более централизованного координатора.