Akka для Java разработчика (часть 1)

В последнее время появилось довольно много языков программирования, которые используют для JVM как платформу для выполнения. Одним из наиболее «горячих» тем для обсуждения в последнее время является Scala. В этой статье я не буду рассматривать заслуженно это или нет, просто хочу рассказать как можно использовать средства этого языка используя Java и не написав ни одной строчки на Scala.

Концепция actor'ов родилась задолго до появления Scala и была реализована на различных языках программирования, в том числе и Java. Если формулировать в двух словах, то actor это объект, который занимается обработкой посылаемых ему сообщений. Важно чтобы сообщения были не изменяемые (immutable) и последовательность их обработки соответствовала порядку их поступления. Собственно это основа концепции, а дальше уже идут особенности реализации — сохраняется ли очередь сообщений для объекта, есть ли возможность поменять код обработки сообщений в процессе работы, привязывается ли actor к потоку или пулу потоков, возможность отправки сообщений на actor'ы на других машинах и т.д.

Вообще если кто знаком с JEE, то actor очень похож на Message bean, только очередь сообщений живет внутри JVM.
Автор языка Scala решил реализовать концепцию actor'ов для языка Scala и назвал ее Akka и развивает этот проект как коммерческий с открытыми исходниками. Проект развивается довольно активно, текущая версия 1.1.3 и идет работа на версией 2.0, которую собираются выпустить осенью. Там будет поддержка распределенных очередей и лучшая организация кластеров.

Почему я выбрал именно Akka? Во-первых хотелось пощупать новый способ борьбы со сложностью распределенных вычислений и high-load систем. Второе и очень важное это документация. Про actor'ы сейчас активно только два проекта JetLang и Akka, и Akka подкупает подробной документацией как для Java, так и для Scala, с примерами кода и подробными объяснениями где и чего. И собственно третья причина, это отсутствие необходимости изучать новый язык для того чтобы получить преимущества использования технологии.

Итак, начнем.

Для того, чтобы писать что-то с использованием Akka необходимо скачать дистрибутив akka.io/downloads или подключить в maven следующие библиотеки.

<repository><br> <id>akka.repository</id><br> <name>Akka Maven Repository</name><br> <url>akka.io/repository</url><br></repository><br><br><dependency><br>  <groupId>se.scalablesolutions.akka</groupId><br>  <artifactId>akka-actor</artifactId><br> <version>1.1.3</version><br></dependency><br><br><dependency><br>  <groupId>se.scalablesolutions.akka</groupId><br>  <artifactId>akka-typed-actor</artifactId><br> <version>1.1.3</version><br></dependency><br><br>* This source code was highlighted with Source Code Highlighter.

В Akka реализованы два типа actor'ов — UntypedActor и TypedActor. UntypedActor предоставляет стандартный интерфейс для обработки сообщений через метод onReceive

import akka.actor.UntypedActor;

public class TestUntypedActor extends UntypedActor {

public void onReceive(Object message) throws Exception {
  if (message instanceof MySuperMessage) {
  System.out.println("We've got super message - " + message);
 } else if (message instance of String) {
  System.out.println("Process string - " + message);
}
}

* This source code was highlighted with Source Code Highlighter.

Таким образом мы получим множество классов с которыми можно взаимодействовать через отправку сообщений.
Для TypedActor необходимо сначала описать интерфейс взаимодействия, а потом его реализовать с помощью actor'а.

interface RegistrationService {
void register(User user, Credentials cred);
User getUserFor(String username);
}

import akka.actor.TypedActor;

public class RegistrationServiceImpl extends TypedActor implements RegistrationService {
public void register(User user, Credentials cred) {
  ... // Регистрация пользователя
}

public User getUserFor(String username) {
  ... // Реализация поиска пользователя
 return user;
}
}

* This source code was highlighted with Source Code Highlighter.

В этом случае у нас будет множество классов с интерфейсами для бизнес-логики и реализацией с помощью actor'ов. И если вы в вашем приложении вы используете Spring, то интегрирование actor'ов в существующие приложение будет на уровне конфигурации и пройдет практически безболезненно (в следующих статьях я покажу как).

Итак у нас есть описания actor'ов и теперь необходимо создать как-то экземпляры классов, потому что простой new тут не работает. Для того чтобы создавать UntypedActor используется фабрика Actors

ActorRef testActor = Actors.actorOf(TestUntypedActor.class);

* This source code was highlighted with Source Code Highlighter.

После создания фабрика возвращает ссылку (ActorRef) на конкретный экземпляр actor'а, которую можно сериализовать или отправить на другую машину для отправки сообщения конкретному экземпляру (в случае remote actor).

Для создания TypedActor используется следующий код

// Первый параметр - интерфейс, второй - реализация
UserService service =
(UserService) TypedActor.newInstance(UserService.class, UserServiceImpl.class);

// Или если у TypedActor нет конструктора без параметров

UserService service = TypedActor.newInstance(UserService.class, new TypedActorFactory() {
public TypedActor create() {
  return new UserServiceImpl("default user storage"));
});

* This source code was highlighted with Source Code Highlighter.

Который возвращает ссылку на интерфейс для общения с actor'ом.

В Akka реализовано три типа отправки сообщений actor'ам — fire-and-forget («отправил и забыл», асинхронная отправка сообщения без ожидания результата), request-reply («вопрос-ответ», отправка сообщения и ожидание ответа, синхронный режим), request-reply-with-future («отправить и получить ответ в будущем», отправка сообщения и получения ответа дальше по коду с помощью специального объекта).

Для UntypedActor все три способа выглядят вот так

actorRef.sendOneWay("Hello"); // Посылает в качестве сообщения строчку

actorRef.sendOneWay(new MySuperObject(), сontext()); // Отправить объект в качестве сообщения. Кроме того будет отправлен context текущего actor'а (из которого происходит вызов), для того чтобы можно было отправить ответ с помощью context().reply()

Object result = actorRef.sendRequestReply("get result"); // Отправить строчку как сообщение и ждать ответ. В методе можно также передать context() и таймаут для ожидания ответа (при истечении таймаута будет ActorTimeoutException)

Future<MyResult> future = actorRef.sendRequestReplyFuture("Hello"); // Отправить сообщение и получить объект для извлечения результатов. Можно передать context() и таймаут. У самого объекта Future есть методы await (ожидание результата), result (получить результат), exception (получить исключение если было)

* This source code was highlighted with Source Code Highlighter.

Для TypedActor способы вызова зависят от сигнатуры метода. Есть метод не возвращает значений значений (void), то неявно используется вызов fire-and-forget. Если возвращается какой-либо объект, то используется request-reply. Если метод возвращает значение Future, то используется request-reply-with-future.

Ну и последнее на сегодня — не забывайте стартовать UntypedActor'ы и останавливать для освобождения ресурсов. Если посылать сообщения не стартовавшему actor'у, то они будут складываться в его очередь и не обрабатываться. Поэтому понять что что-то не работает будет тяжело.