Реактивные акторы на java

[Googolplex]

Спасибо за статью!

Я бы посоветовал переписать её с использованием Java 8 (вот здесь ещё информация). Код станет гораздо более красивым и понятным.

[harati]

Хорошее замечание, когда соберётся ещё несколько пожеланий по добавлению интересующих вещей — так и сделаю.

[daron666]

Да! Хочется продолжения именно с лямбдами. Сейчас не так много материала по этой теме.

[knott]

Где подписаться?

[gothica]

А есть ли готовый пример, например, на гитхабе?

[bvitaliyg]

Великолепная статья, давно хотелось приобщиться к акторам и реактивному программированию.

[Regis]

Сообщение — любой неизменяемый объект реализующий интерфейс Serializable.

Неверно. В общем случае, сообщение — это любой объект. Обычно по-хорошему, он должен быть эффективно неизменяемым (immutable), но тут без принуждения. Serializable же требуется, если вы хотите использовать remoting, но вам лень делать нормальную сериализацию и вы хотите ограничиться дефолтной java-вской встроенной.

ActorRef — ссылка на конкретный, гарантированно существующий актор, аналог StrongReference. Отправляя по ней сообщение существуют гарантии, что оно дойдёт, или же будет брошен exception.

Неверно. Актор уже вполне может быть остановлен и уничтожен, а ActorRef у вас останется на руках. Соответственно никаких гарантий, что актор получит данное сообщение. PS: Привет DeadLetters.

Сразу в глаза бросаются некоторая расточительность — мы создаем акторы для одной операции и потом их уничтожаем.

С точки зрения общей концепции Акки — во многих ситуациях это вполне себе нормально.

Теперь займёмся кодом ядра, оно выполняет две функции — распределяет работу и объединяет результаты.

На практике лучше так не делать. Лучше по-возможности разделять задачи акторов. Пока вы одновременно обрабатываете ровно один запрос на поиск простых чисел, то и так сойдет. Но вы ведь не для этого решили использоваться Akka, верно? Поэтому «агрегатор результатов» лучше создавать новый под каждый запрос на поиск простого числа.

Роутер — специализированный объект, передающий входящие сообщения акторам, используя определённую стратегию их выбора. Бывает двух типов — пул и группа.

«пул и группа» — это вы про разные способы инициализации роутеров?

Очевидно, что происходит при попытке актора вызвать блокирующую операцию — поток блокируется, и остальные акторы начинают обрабатываться куда медленнее. А если количество заблокированных акторов становится = количеству потоков то система замирает. Даже из такой неловкой ситуации есть выход — Future.

Эм. Furure на самом деле немного о другом. Более того, ваш пример их использования совершенно не помогает для решения проблемы блокировки потоков. Если вы делаете блокирующую операцию внутри Future, то вы по-прежнему блокируете под это поток из общего пула потоков (ну или из того, который вы указали). Документация Akka описывает совсем другой рецепт решения: деление всех акторов на две группы (можно больше). Группа с «правильными» акторами, которые никогда не блокируют. В идеале — совсем без какой-либо синхронизации в каком-либо виде. И группа с «меделенными» акторами, которая может блокировать, но имеет свой отдельный выделенный и явно ограниченный пул потоков. Чаще всего — много пулов. Например пул акторов, которые работают с синхронной БД; пул для работы с сетью и т.д.

[harati]

Сообщение — любой неизменяемый объект реализующий интерфейс Serializable.

Неверно. В общем случае, сообщение — это любой объект. Обычно по-хорошему, он должен быть эффективно неизменяемым (immutable), но тут без принуждения. Serializable же требуется, если вы хотите использовать remoting, но вам лень делать нормальную сериализацию и вы хотите ограничиться дефолтной java-вской встроенной.

Концепция акки строго запрещает изменяемые сообщения, только immutable. Просто нет средств языка гарантировать эту неизменяемость, посему введено соглашение.

ActorRef — ссылка на конкретный, гарантированно существующий актор, аналог StrongReference. Отправляя по ней сообщение существуют гарантии, что оно дойдёт, или же будет брошен exception.

Неверно. Актор уже вполне может быть остановлен и уничтожен, а ActorRef у вас останется на руках. Соответственно никаких гарантий, что актор получит данное сообщение. PS: Привет DeadLetters.

Хорошее уточнение, спасибо, исправлю.

Теперь займёмся кодом ядра, оно выполняет две функции — распределяет работу и объединяет результаты.

На практике лучше так не делать. Лучше по-возможности разделять задачи акторов. Пока вы одновременно обрабатываете ровно один запрос на поиск простых чисел, то и так сойдет. Но вы ведь не для этого решили использоваться Akka, верно? Поэтому «агрегатор результатов» лучше создавать новый под каждый запрос на поиск простого числа.

Безусловно, задачи разделять нужно, но в данном примере есть очередь задач с счётчиками, иллюстрируя варианты (кроме тривиальных, вроде создания отдельного актора ради одного запроса) решения. К примеру, если идёт обработка отдельных пакетов (игровая логика, HTTP) — создание актора для каждого уже расточительно, и умение находить эту грань очень важно.

Очевидно, что происходит при попытке актора вызвать блокирующую операцию — поток блокируется, и остальные акторы начинают обрабатываться куда медленнее. А если количество заблокированных акторов становится = количеству потоков то система замирает. Даже из такой неловкой ситуации есть выход — Future.

Эм. Furure на самом деле немного о другом. Более того, ваш пример их использования совершенно не помогает для решения проблемы блокировки потоков. Если вы делаете блокирующую операцию внутри Future, то вы по-прежнему блокируете под это поток из общего пула потоков (ну или из того, который вы указали). Документация Akka описывает совсем другой рецепт решения: деление всех акторов на две группы (можно больше). Группа с «правильными» акторами, которые никогда не блокируют. В идеале — совсем без какой-либо синхронизации в каком-либо виде. И группа с «меделенными» акторами, которая может блокировать, но имеет свой отдельный выделенный и явно ограниченный пул потоков. Чаще всего — много пулов. Например пул акторов, которые работают с синхронной БД; пул для работы с сетью и т.д.

Да, как вы верно указали — Future блокируется на потоке из пула, который мы указали. Именно для этого создаётся отдельный пул с множеством процессов, а Future — это инкапсулирует до уровня логики акки. В разделе о контексте выполнения я об этом говорил.

[Regis]

Концепция акки строго запрещает изменяемые сообщения, только immutable. Просто нет средств языка гарантировать эту неизменяемость, посему введено соглашение.

В идеальном случае — да, хорошо бы, чтобы все сообщения были Immutable. Но в реальной жизни у этого есть своя цена. На самом деле реальное условие таково: объект, передаваемый в качестве сообщения должен быть «корректно виден» (т.е. соблюдается явный happens before для всех изменений) вне зависимости от того, в каком из потоков к нему идут оращения, так как каждое сообщение может обрабатываться каждым актором в произвольном потоке из пула. Но объестение, что такое «корректно виден» и в каких случаях это условие выполняется, а в каких нет — довольно обширная и сложная тема. Поэтому авторы документации к Акке пишут, что нужен Immutable объект. Это просто и понятно. Но более жестко, чем на самом деле требуется.
И опять же, Serializable тут не при чем.

К примеру, если идёт обработка отдельных пакетов (игровая логика, HTTP) — создание актора для каждого уже расточительно, и умение находить эту грань очень важно.

По поводу «создание актора для каждого уже расточительно». Если вы про создание актора под отдельный http-запрос или каждую команду с клиента, то мой ответ: нет, не расточительно. Аргументы:
1. Это вполне рекомендованный паттерн, описанный в документации Акки.
2. Команда разработчиков Акки от версии к версии постоянно следит, чтобы создание актора было быстрой и дешевой операцией. Не зря они регулярно выкладывает бенчи с миллионами акторов. Да и в своей статье вы это сами же упоминаете.
3. Тот самый Patterns.ask для Future, который вы упоминаете в статье, делает ни что иное как создает новый актор, который предназначен только лишь для того, чтобы получить единственный ответ и тут же умереть. По вашей логике выходит, что если вспомогательный актор создается через Patterns.ask, то это можно. А если то же самое делать ручками, то нельзя. Смахивает на двойные стандарты :) Note: да, я в курсе, что временный актор, создаваемый Patterns.ask немного «дешевле» обычного. Но совсем не на много.
4. Чтобы создание акторов стало узким местом вашего приложения — нужно очень постараться. Скорее вы упретесь в выделение памяти под immutable сообщения, чем в создание акторов. Это замечание взято не «с потолка», а на базе двухлетнего опыта разработки высокопроизводительного игрового сервера с использованием Акки.

Да, как вы верно указали — Future блокируется на потоке из пула, который мы указали. Именно для этого создаётся отдельный пул с множеством процессов, а Future — это инкапсулирует до уровня логики акки. В разделе о контексте выполнения я об этом говорил.

Я о том, что вы говорите про проблему блокировки и предлагаете её решать с помощью Futures. Но сами Futures совершенно не помогают в проблеме блокировки потоков. Если же вы предлагаете использовать Futures только для того, чтобы выполнять их в другом контексте выполнения, то:
1. Для того, чтобы распределить выполнение по разным контекстам, можно задать контексты выполнения для самих акторов. Futures для лишь этого — не нужны.
2. Вы просто переложили блокировки из одного пула в другой, без явного критерия, что можно блокировать, а что нет, без явной логики такого разделения. Вместо стагнации акторов целиком, у вас будет стагнация частей логики акторов (в виде Futures). Нередко это может быть даже хуже (можно получить что-то вроде live lock, только вместо потоков с конкуренцией за общие ресурсы здесь будет конкуренция акторов за общие потоки).

[CheatEx]

Вы ещё вордкаунта не хватает, а то наверно не все прочувтвовали достоинтства акторов для параллельного программирования…

[ImLiar]

Все таки в scala акторы на akka смотрятся куда как приятнее. Тем более, что иммутабельность сообщений в java — та еще боль ниже пояса, а в scala связка case class/pattern matching выглядит именно так, как это должно быть.