{"blocks":[{"type":"expertLegacy","data":{"id":"1375"}},{"type":"paragraph","data":{"text":"Akka была создана для решения проблем в высоконагруженных системах. Делается это с помощью модели акторов."}},{"type":"paragraph","data":{"text":"Актор сам по себе — это сущность, которая получает некий Message, обрабатывает его атомарно и посылает сообщение следующему актору. Акторы работают асинхронно, не передают контекст приложения между собой и не имеют прямого обращения к экземпляру другого актора. Каждый актор ничего не знает о другом акторе, у него есть только некий адрес. Общение происходит исключительно через отправку Message. Сами Messages должны быть immutable."}},{"type":"paragraph","data":{"text":"Также у актора есть Mailbox – это очередь входящих сообщений. Операция отправки сообщения является неблокирующей. Актор просто отправил сообщение и ничего не ждет в ответ."}},{"id":"40aecdba-6ef7-4323-8a16-be2534065542","type":"banner-blank","data":{}},{"type":"paragraph","data":{"text":"В нашем приложении будет иерархия акторов – главный актор-супервизор, который будет создавать и инициализировать дочерние акторы-воркеры. Такая система с супервизором гарантирует, что в случае сбоя в работе супервизора по умолчанию он будет перезапускаться сам и создавать заново все дочерние акторы, начинать с чистого листа. Также у нас будет три дочерних актора:"}},{"type":"list","data":{"items":["Consumer– актор, который будет читать сообщения из Kafka в формате protobuf и посылать эти файлы следующему актору – FileSaver. Сообщения представляют собой файл, который содержит какой-то набор полей, в том числе список неких messages. Messages из файла тоже содержат какой-то набор полей. Все эти данные нам необходимо сохранить в соответствующие таблицы в БД.","FileSaver – актор, который будет получать сообщения от консюмера, сохранять данные в БД в таблицу file и посылать сообщения с вложенными в файл message следующему актору – MessageSaver.","MessageSaver – актор, который будет получать сообщения от fileSaver’а и сохранять их в БД в таблицу message."],"style":"ordered"}},{"type":"header2","data":{"level":2,"text":"Первоначальная настройка"}},{"type":"paragraph","data":{"text":"Итак, прежде всего подключим в Maven артефакты akka-actor_2.12 и akka-contrib_2.12 – это основной набор инструментов Акка для создания приложения:"}},{"type":"code","data":{"code":"\r\n com.typesafe.akka\r\n akka-actor_2.12\r\n ${akka.version}\r\n \r\n \r\n com.typesafe.akka\r\n akka-contrib_2.12\r\n ${akka.version}\r\n \r\n \r\n com.typesafe.akka\r\n akka-slf4j_2.12\r\n ${akka.version}\r\n \r\n \r\n org.scala-lang.modules\r\n scala-java8-compat_2.12\r\n ${scala-java8-compat.version}\r\n ","language":"xml lazy-code"}},{"type":"paragraph","data":{"text":"Для старта приложения нам потребуется ActorSystem. Он нужен для того, чтобы создать контекст нашего приложения. Из него будет создан наш первый и главный актор-супервизор."}},{"type":"paragraph","data":{"text":"Создать ActorSystem можно с помощью статичного метода ActorSystem.create(). В него можно передать имя нашей системы и конфигурацию. Конфигурацию мы будем хранить в ресурсах в файле application.conf, именно он будет по умолчанию парситься при загрузке контекста. Полный код конфигурации можно посмотреть на GitHub, приведу тут только основные настройки:"}},{"type":"paragraph","data":{"text":"Пулы акторов:"}},{"type":"code","data":{"code":"deployment {\r\n /applicationSupervisor/fileSaver {\r\n router = round-robin-pool\r\n nr-of-instances = 10\r\n }\r\n /applicationSupervisor/messageSaver {\r\n router = round-robin-pool\r\n nr-of-instances = 10\r\n }\r\n }","language":"php lazy-code"}},{"type":"paragraph","data":{"text":"Данный кусок конфигурации говорит о том, чтобы при создании акторов с именами fileSaver и messageSaver они будут созданы с балансировщиком с политикой round-robin-pool, на котором будет крутиться по 10 инстансов каждого актора."}},{"type":"paragraph","data":{"text":"Диспетчер:"}},{"type":"code","data":{"code":"dispatcher {\r\n type = PinnedDispatcher\r\n executor = \"thread-pool-executor\"\r\n }","language":"php lazy-code"}},{"type":"paragraph","data":{"text":"Тут указывается, на каком диспетчере будет выполняться данный актор. Эта настройка не является обязательной. В нашем простом приложении мы не будем её добавлять, а оставим это по умолчанию."}},{"type":"header2","data":{"level":2,"text":"Создание актора-супервизора и дочерних акторов"}},{"type":"paragraph","data":{"text":"Мы создали систему. Теперь мы можем попросить систему создать главный актор-супервизор с помощью метода system.actorOf();. Этот метод ожидает от нас конфигурацию актора (Props) и его имя. Если с именем всё понятно, то про конфигурацию надо сказать отдельно."}},{"id":"ee72e28c-e3b1-4903-966d-2774f208beb4","type":"banner-blank","data":{}},{"type":"paragraph","data":{"text":"Для начала нужно создать класс супервизор, который будет являться актором. Создадим класс ApplicationSupervisor и наследуемся от AbstractActor.
Для создания конфигурации этого актора нужно описать статичный метод props(), который будет создавать и возвращать экземпляр конфигурации этого класса. Пример кода ниже."}},{"type":"code","data":{"code":"public static Props props() {\r\n return Props.create(ApplicationSupervisor.class, ApplicationSupervisor::new);\r\n }","language":"clike lazy-code"}},{"type":"paragraph","data":{"text":"Этого уже достаточно, чтобы создать актор-супервизор и запустить приложение. Вот пример кода для старта метода main."}},{"type":"code","data":{"code":"public final class Main {\r\n\r\n private static final Logger LOG = LoggerFactory.getLogger(Main.class);\r\n private static final String SERVICE_NAME = \"simple-example\";\r\n\r\n private Main() {}\r\n\r\n /**\r\n * Main entry point to application.\r\n *\r\n * @param args cmd args.\r\n */\r\n public static void main(String[] args) {\r\n LOG.info(\"Starting application\");\r\n final ActorSystem system = ActorSystem.create(\r\n SERVICE_NAME, ConfigFactory.load().getConfig(SERVICE_NAME));\r\n system.actorOf(ApplicationSupervisor.props(), \"applicationSupervisor\");\r\n }\r\n}","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"После того как мы унаследовали AbstractActor, необходимо переопределить метод createReceive(), который будет обрабатывать входящие сообщения. Сам актор-супервизор не будет получать никаких сообщений. Он будет отвечать только за инициализацию DataSource, которые необходимы для работы с БД и инициализацию остальных дочерних акторов. Соответственно, createReceive() в случае получения любого сообщения будет логировать его и ничего не делать дальше."}},{"type":"code","data":{"code":"@Override\r\n public Receive createReceive() {\r\n return receiveBuilder()\r\n .matchAny(m -> LOG.warn(\"Unknown message: {}\", m))\r\n .build();\r\n }\r\n \r\n @Override\r\n public void postStop() throws Exception {\r\n super.postStop();\r\n PostgreSqlUtils.terminate(this.dataSource);\r\n LOG.info(\"ApplicationSupervisor stopped\");\r\n }","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"Теперь нам нужно создать дочерних акторов fileSaver и messageSaver. Для этого вызовем метод-фабрику context().actorOf(), который ждет от нас конфигурацию актора и наименование. В данном случае нам уже нужен не один актор, а целый пул. Такую конфигурацию можно создать с помощью метода FromConfig.getInstance().props(). Это означает, что конфигурация пула будет взята из созданного ранее application.conf, который мы описали выше."}},{"type":"paragraph","data":{"text":"В конце создадим актор Consumer. Он будет один, поэтому для него конфигурацию пула передавать не надо. Иерархия у нас такая, что Consumer читает Kafka и отправляет сообщения в fileSaver. Тот, в свою очередь, сохраняет в БД данные и отправляет сообщения в messageSaver, который сохраняет в БД данные. Соответственно, создавать эти акторы надо в обратной последовательности, чтобы передавать ссылку в нужной иерархии."}},{"type":"code","data":{"code":"private void initActors() {\r\n final FileDao fileDao =\r\n new FileDao(this.dataSource);\r\n final MessageDao messageDao =\r\n new MessageDao(this.dataSource);\r\n\r\n ActorRef messageSaver = context().actorOf(\r\n FromConfig.getInstance().props(MessageSaver.props(\r\n messageDao\r\n )),\r\n \"messageSaver\"\r\n );\r\n ActorRef fileSaver = context().actorOf(\r\n FromConfig.getInstance().props(FileSaver.props(\r\n messageSaver,\r\n fileDao\r\n )),\r\n \"fileSaver\"\r\n );\r\n\r\n context().actorOf(Consumer.props(fileSaver), \"consumer\");\r\n }","language":"java lazy-code"}},{"type":"header2","data":{"level":2,"text":"Отправка и обработка сообщений"}},{"type":"paragraph","data":{"text":"Теперь, когда мы создали акторов, давайте обратим внимание на работу первого в очереди – это consumer. Переопределенный метод preStart() будет инициализировать подписку на некий кафка-топик и вычитывать первое сообщение из него. Далее он будет отправлять сообщение самому себе."}},{"type":"paragraph","data":{"text":"Чтобы отправить сообщение актору, нам нужна ссылка на него. В данном случае мы отправляем сообщение себе, поэтому ссылку на самого себя возвращает метод self(). Далее мы вызываем метод tell(), в который передаём сообщение и ссылку на актор-отправитель."}},{"type":"code","data":{"code":"self().tell(new KafkaBatch(batch), self());","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"Для обработки входящих сообщений переопределяем метод createReceive() и в нём создаём receiveBuilder(), который будет матчить входящие сообщения на соответствующие методы для их обработки."}},{"type":"code","data":{"code":"@Override\r\n public Receive createReceive() {\r\n return receiveBuilder()\r\n .match(KafkaBatch.class, this::processKafkaBatch)\r\n .match(Done.class, m -> handleDone(m.getId()))\r\n .matchAny(m -> LOG.warn(\"Unknown message: {}\", m))\r\n .build();\r\n } \r\n\r\n /**\r\n * Process message from kafka.\r\n *\r\n * @param kafkaBatch byte array with message from kafka\r\n */\r\n private void processKafkaBatch(final KafkaBatch kafkaBatch) {\r\n currentBatchId = UUID.randomUUID().toString();\r\n final List records = new LinkedList<>();\r\n for (ConsumerRecord data : kafkaBatch.getEvents()) {\r\n try {\r\n final ConsumerData.File file =\r\n ConsumerData.File.parseFrom(data.value());\r\n records.add(new FileWrapper(currentBatchId, file));\r\n } catch (InvalidProtocolBufferException e) {\r\n LOG.error(\"Unknown data in Kafka message: {}\", data.value());\r\n }\r\n }\r\n if (records.isEmpty()) {\r\n commitAndAskForNewData();\r\n } else {\r\n amountDone = 0;\r\n currentBatchSize = records.size();\r\n records.forEach(file -> fileSaver.tell(file, self()));\r\n }\r\n }","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"В методе processKafkaBatch() мы запоминаем новый BatchId, он понадобится нам позже. Далее парсим протобаф-файл и создаем список обёрток, которые содержат файл и BatchId. Далее мы запоминаем количество получившихся сообщений, указываем amountDone = 0 (это говорит о том, что количество сообщений об успешном завершении работы вложенных акторов сейчас = 0) и в цикле отправляем по одному файлу следующему актору на обработку fileSaver.tell(file, self())."}},{"type":"paragraph","data":{"text":"Зачем нам нужна эта часть с BatchId и amountDone? За тем, что мы хотим знать, когда fileSaver выполнил свою работу и нам можно коммитить и получать из Kafka новый Batch. Для этого fileSaver после сохранения в БД данных и отправки сообщений дальше в messageSaver будет отправлять назад консюмеру Done сообщение с текущим BatchId. Метод для обработки входящих Done следующий:"}},{"type":"code","data":{"code":"public void handleDone(String batchId) {\r\n if (Objects.equals(batchId, currentBatchId)) {\r\n amountDone++;\r\n if (amountDone == currentBatchSize) {\r\n commitAndAskForNewData();\r\n }\r\n }\r\n }","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"Вот так будет выглядеть обработка входящих в акторе FileSaver:"}},{"type":"code","data":{"code":"@Override\r\n public Receive createReceive() {\r\n return receiveBuilder()\r\n .match(FileWrapper.class, this::processfile)\r\n .match(Done.class, m -> handleDone(m.getId()))\r\n .matchAny(m -> LOG.warn(\"Unknown message: {}\", m))\r\n .build();\r\n }\r\n\r\n private void processfile(final FileWrapper fileWrapper) {\r\n sender = sender();\r\n receivedBatchId = fileWrapper.getBatchId();\r\n currentBatchId = UUID.randomUUID().toString();\r\n long fileId = 0;\r\n ConsumerData.File file = fileWrapper.getFile();\r\n final List messagesList = file.getMessagesList();\r\n try {\r\n fileId = fileDao.saveFile(file);\r\n } catch (SQLException e) {\r\n LOG.error(e.getMessage());\r\n }\r\n if (messagesList.isEmpty()) {\r\n sender().tell(new Done(fileWrapper.getBatchId()), self());\r\n } else {\r\n amountDone = 0;\r\n currentBatchSize = messagesList.size();\r\n long finalFileId = fileId;\r\n messagesList.forEach(message ->\r\n messageSaver.tell(new MessageWrapper(\r\n currentBatchId,\r\n finalFileId,\r\n message\r\n ), self())\r\n );\r\n }\r\n }","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"Здесь мы запоминаем ссылку на sender, она нам понадобится позже. А дальше делаем по аналогии с консюмером. Сохраняем в БД нужные данные и, если вложений больше нет, то говорим сендеру Done сразу, а если вложения есть, то посылаем их по одному актору messageSaver. Когда messageSaver вернёт нам соответствующее количество Done-сообщений, мы тоже говорим сендеру Done."}},{"type":"code","data":{"code":"public void handleDone(String batchId) {\r\n if (Objects.equals(batchId, currentBatchId)) {\r\n amountDone++;\r\n if (amountDone == currentBatchSize) {\r\n sender.tell(new Done(receivedBatchId), self());\r\n }\r\n }\r\n }","language":"java lazy-code"}},{"type":"paragraph","data":{"text":"На этом разработка нашего простого приложения закончена. Мы не обрабатываем ошибочные ситуации в процессе работы актора, например, если произошёл какой-то SQLError, то мы его просто логируем. Но можно, например, пробрасывать Fail-сообщение отправителю вверх, и он уже будет его обрабатывать каким-то специфичным образом."}},{"id":"7c9d5e03-8928-4de5-9acd-78a6dc01eb1b","type":"banner-blank","data":{}},{"type":"paragraph","data":{"text":"В результате у нас получилась такая конвейерная система, которую легко масштабировать как вертикально, так и горизонтально. У нас есть устойчивость к сбоям."}},{"type":"paragraph","data":{"text":"Мы можем добавить мониторинг на каждую из стадий, чтобы понимать, где и кто тупит, чтобы потом масштабировать отдельный участок с помощью увеличения пула или разбития на несколько отдельных стадий и т.д. Само по себе создание актора это очень легковесный процесс. Создать или уничтожить его легко, это никак не связано с созданием или уничтожением нитей."}},{"type":"paragraph","data":{"text":"В завершении хотелось бы отметить, что приложение с использованием Akka легко проектировать и разрабатывать. Также у него есть очень удобный фреймворк для тестирования. Так что при выборе решения для разработки высоконагруженного приложения Akka даёт полный набор инструментов для его создания."}}]}

Ошибка в настройках сайта