Тёмные глубины многопоточности

 Почти год назад делал тестовое задание для одной компании. Но тогда проходил собеседования сразу в несколько контор и было довольно много работы и домашних дел. Плюс, решил в своём решении показать сразу всё, на что был способен и сильно переусложнил его для тестового задания. Поэтому сделать тогда задание за два выходных дня не успел. Застрял на дедлоках и не смог быстро найти места, где они возникают.

То решение было стыдно показывать ревьюеру. Но ещё хуже было не показать ничего. Поэтому у меня весь год “чесались руки” сделать его как следует. И вот, наконец, этот час настал.
Решение получилось без изысков. Я знаю, что многое в нём можно доделать и улучшить. Но, выполняя задачу, своей цели я достиг, и решение на правильных данных работает быстро и считает, как надо.

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

Возвращаюсь к теме статьи. Если коротко, то задание звучало так: написать программу, которая будет парить данные из файла и для разных данных (разных инструментов сделок) считать разные статистики. Текст задания можно посмотреть в моём GitHub-репозитории.

Решение задачи состоит из следующих частей:
- парсер файлов. Как понятно из названия, его задача - открыть файл и распарсить его содержимое;
- модель - представляет собой распарсенные данные;
- очереди - содержат распарсенные данные для разных инструментов (число очередей равно числу инструментов, для которых необходимо посчитать статистику);
- калькуляторы статистики - рассчитывают статистику для разных инструментов сделок;
- класс, содержащий основную связующую логику приложения.

Рассмотрим их более подробно:
Processor.
Класс с основной логикой приложения. В нём мы создаём объекты классов нашего приложения, очереди для данных для калькуляторов, защёлку CountDownLatch и пул потоков, в которых параллельно будут исполняться: парсер и калькуляторы. После завершения выполнения своих задач парсером и калькуляторами, производится вывод результатов в консоль.

Parser.
Построчно считывает данные из файла и складывает их в очереди. После завершения своей работы, помещает в каждую очередь маркерную модель, которая будет сигнализировать потребителям данных, что новых даных не будет и пора завершать работу. Также  по завершении своей работы Parser открывает защёлку CountDownLatch - это сигнал для начала работы калькулятором, считающим статистику по десяти самым последним по дате данным инструмента №3.

Очереди.
Две очереди являются экземплярами LinkedBlockingQueue.
Если при вызове метода put() такой очереди она была заполнена полностью, выполнение метода блокируется в ожидании, когда в очереди появится свободное место.
Если при вызове метода take() такой очереди она была пуста, выполнение метода блокируется в ожидании, когда в очереди появится хотя бы один элемент.
Очередь BoundedPriorityBlockingQueue является кастомной реализацией блокирующей очереди с приоритетом. Её задача - хранить в себе только 10 записей. Приоритет на помещение в эту очередь имеют последние по дате данные для инструмента №3. Приоритет определяется с помощью метода  compareTo(), реализованного в модели Quotation.

Калькуляторы.
AverageValueByMonthCalculator - считает среднее значение цены за определённый месяц.
AverageValueCalculator - считает среднее значение цены за всё время.
LastTenSumCalculator - считает сумму последних десяти по дате цен.
Калькуляторы считают данные каждый из своей очереди. И прекращают работу после получения из своих очередей маркерных данных.
Калькулятор LastTenSumCalculator начинает свои расчёты после того, как Parser опустит защёлку CountDownLatch, что будет означать, что в очереди лежат последние 10 по дате данные для инструмента №3 и помещение туда новых данных не предвидится.

Благодаря такой реализации, парсинг данных и подсчёт статистики производятся параллельно и практически одновременно.
Потоковая обработка данных при помощи блокирующих очередей защищает приложение от переполнения памяти. Файл для парсинга может быть какого угодно большого размера.
Завершение парсером работы путём помещения в очередь маркерных данных, защищает приложение от live lock.

Apache Ignite cache: добавляем конфигурацию для логгера

 Итак, наш сервис стартует в окружении Apache Ignite Service Grid, загружает файл с данными и парсит данные в Кеш.
Что дальше?
Давайте, попробуем настроить вывод логов нашего сервиса в отдельный файл, чтобы они не смешивались с логами кластера Ignite и мы могли в любой момент быстро их просмотреть.
Для этого нам понадобятся две вещи:
1) Создать файл с конфигурацией для логгера.
2) Дать понять нашему сервису, что писать логи он теперь должен в новый логгер.

По первому пункту - был создан файл ignite-logback.xml
. В нём были заданы параметры для сохранения логов в обычный лог-файл, в json-файл и в консоль.
По второму пункту - добавим в метод инициализации нашего сервиса следующий код:

Здесь мы получаем контекст логгера, сбрасываем его конфигурацию, которая была подтянута в момент старта окружения Apache Ignite, и указываем конфигуратору путь к нашему файлу с конфигурации логгера - ignite-logback.xml.

Стартуем наш сервис и видим следующие логи:

Которые имеют следующее содержание:
log-файл:

Json-файл:

Снова к теме торгов по зарплате

Уж сколько писано-переписано статей, снято-переснято роликов на ютюбе, а с завидным постоянством находятся те, кто "не знает". Причём, иногда складывается впечатление, что это одни и те же люди гоняют у себя в голове по кругу одну и ту же навязчивую идею. Звучит эта идея так:

"Я не способен. Я хуже других. Но я тоже хочу. Я обхитрю систему. Предложу платить мне меньше, чем другим. Меня с радостью возьмут на работу. А потом я докажу! Я покажу! Заслужу! И мне подымут. Мне обязательно подымут зарплату. А до тех пор нужно будет потерпеть на хлебе и воде. Я смогу!"

Что не так в этой навязчивой идее:

1) Начнём с: "Я не способен. Я хуже других. Но я тоже хочу. Я обхитрю систему. Предложу платить мне меньше, чем другим. Меня с радостью возьмут на работу".

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

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

Поэтому мысль о том, что "Я не способен. Я хуже других. Но я тоже хочу. Я обхитрю систему. Предложу платить мне меньше, чем другим. Меня с радостью возьмут на работу." - логически неверна.

Если тебя берут на работу, то не потому, что твой труд стоит на 10 - 50 % меньше, чем у других кандидатов, а, в первую очередь, потому, что ты можешь справиться с задачей. И только во вторую очередь потому, что твои запросы могли оказаться ниже, чем у других кандидатов.

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

Поскольку мы выяснили, что ты не хуже других кандидатов, то почему ты решил, что должен зарабатывать меньше, чем они? Потому что ты сам себя обманул, поверив мысли о том, что тебя возьмут на работу только за то, что ты согласишься работать за еду.

2) "А потом я докажу! Я покажу! Заслужу! И мне подымут. Мне обязательно подымут зарплату." - да это курам на смех!

Если бы это работало, то достаточно было бы на собеседовании произнести только эти три магические фразы и тебя сразу бы взяли на работу в любую компанию. Но все мы знаем, что это не так.

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

То, что ты прошёл собеседование и тебя приняли на работу, подтверждает тот факт, что ты способен выполнять нужную работодателю работу. А если ты способен выполнять её наравне со всеми, то почему ты должен получать за это меньшую зарплату?

3) "А до тех пор нужно будет потерпеть на хлебе и воде. Я смогу!" - ага... знаешь, эти парни говорят, что готовы жить в Москве (которая для них не родной город) полгода на зарплату 30 тысяч рублей в месяц...

Не лучше ли приложить эти усилия в более конструктивное русло, чем в борьбу за выживание? Зачем истязать свои тело и дух, когда можно хорошо подготовиться к собеседованию, пройти его и получить зарплату, соответствующую рыночному уровню, которая позволит тебе и твоей семье жить не за чертой бедности?

Не оценивай себя низко. Оценивай адекватно рынку. Так и тебе лучше будет и другим.

Как можно продолжать мечтать о зарплате в 60 тыс.руб./мес., когда рубль в цене упал в 1,5 раза?..

Хочешь зарплату на уровне рынка? Подготовься по теории так, чтобы от зубов отскакивало и смело пиши в резюме заветные цифры.

В противном случае, возможны ситуации, вплоть до анекдотических, когда в САМОМ КРУПНОМ БАНКЕ (в котором все деньги лежат) на последнем этапе собеседования тебе скажут: мы тебе не можем дать рыночную зарплату, т.к. у нас ребята с 5-летним стажем получают меньше - обидятся.
Анекдотичность ситуации не в том, что БАНК "не может", а в том, что парни с 5-летним стажем работы в БАНКЕ зарабатывали меньше, чем я, имеющий в 2 раза меньший трудовой стаж, считал нижней планкой на тот момент. Ладно, я - найду работу в другом месте с зарплатой на уровне рынка. Но они-то себя за что наказывают?

Не хочешь быть одним из тех парней? Не демпингуй.

Apache Ignite cache: Запускаем сервис в Apache Ignite Service Grid. Часть 3

 И вот, наконец, мы добрались до файла ignite-config.xml.

Буду разбирать его по кусочкам, т.к. целиком в один скриншот он вряд ли влезет. Да и объяснять, где в нём что, будет проще на небольших кусочках. Полностью этот файл можно будет посмотреть в репозитории проекта.

Объявляем схемы, которые будем использовать в нашем конфиге. Как видите, используются схемы Spring framework. Сам Apache Ignite работает на Spring. И в его конфигурационном файле можно использовать все те же инструменты, что и при XML-конфигурировании Spring-бинов. Это очень большой плюс, т.к. существенно расширяет наши возможности по переносу всей конфигурации Ignite во внешний файл.

Добавляем в переменные окружения флаг, указывающий нодам использовать для сетевого взаимодействия протокол IPv4. Согласно документации Apache Ignite, с протоколом IPv6 кластер работает нестабильно. В продакшене нам нужна стабильность. Поэтому наш выбор - IPv4.

Объявляем бин с конфигурацией ноды.

Отключаем обмен классами между нодами по протоколу P2P. Согласно документации Apache Ignite, такой обмен классами, позволяет нодам обмениваться друг с другом классами. Как показывает практика, работает эта штука не для всех случаев и несколько замедляет работу кластера. Например, я из описания этой фичи понял, что можно копировать наш сервис в директорию libs не всех нод, а только одной. И она уже перешлёт его классы остальным нодам. Но у меня это не заработало и пришлось синхронизировать директории libs на всех нодах самстоятельно. Есть всего один случай, когда PeerClassLoading  мне пригодился, но о нём я расскажу позднее, когда буду писать внешнего клиента для нашего кеша.

Т.к., кроме стабильности, в продакшене важна также скорость работы, отключаем PeerClassLoading.

Конфигурируем TcpDiscoverySpi. Здесь мы говорим кластеру, что все ноды, входящие в кластер, ему нужно искать в списке addresses. Согласно документации, если использовать настройки по умолчанию, то каждая нода будет искать другие ноды по всему диапазону доступных адресов. Когда нод в кластере много, то время, необходимое для запуска кластера, может существенно возрасти.

Кроме того, при дефолтных настройках в кластер могут попасть посторонние ноды, относящиеся к другому кластеру.

Голосуя в продакшене за стабильность и скорость, указываем нашему кластеру из одной ноды искать остальные ноды (которых нет) только на localhost.

Конфигурируем наши сервисы. У нас их два: ReviewRepository и StartService. Указываем:

- name - как они будут называться;

- maxPerNodeCount - количество экземпляров сервиса, которое будет запущено на каждой ноде;

- totalCount - количество экземпляров сервиса, которое будет запущено во всём в кластере. Эта настройка есть только у StartService и равна 1. Это значит, что такой сервис будет запущен в единственном экземпляре на весь кластер (cluster singleton). У ReviewRepository такой настройки нет и он будет запущен в одном экземпляре на каждой ноде;

- nodeFilter - указываем,по какому признаку кластер должен выбрать ноды, на которых будет запущен сервис. В нашем случае это должны быть ноды с ролью  "service-node".

Конфигурируем наши кеши. У нас их два: один - для регистрации активности приложения, второй - для сохранения распарсенных данных. Задаём следующие параметры:

- name - имя кеша. В нашем случае это: activity_cache  и review_cache. Этим именам должны соответствовать значения, которые мы добавили в наши PROPERTIES  в классе ru.emi.ignitecache.utility.PropertyUtility нашего сервиса:

- cacheMode - поведение кеша. У нас это PARTITIONED - кеш будет разделён на части между нодами. Если будет три ноды, то на каждой из них будет находиться примерно 1/3 даных кеша;

- backups - количество резервных копий кеша. У нас их по одной на каждый кеш;

- statisticsEnabled - будет сервис отдавать метрики или нет. Третий столп, на котором держится продакшен - метрики. Поэтому ставим true.

Здесь мы указываем, что наша нода будет выполнять роль "service-node", т.е., что на ней должны быть запущены наши сервисы StartService и ReviewRepository.

Здесь я выделил кешу 4 ГБ памяти. Сделал это потому, что при больших объёмах данных Apache Ignite падал с ошибкой, говорящей о том, что ему не хватило памяти. Желающие более досконально разобраться с этим параметром, могут найти все подробности в документации Apache Ignite.

Здесь мы объявляем бины наших сервисов (см. строчки 58 и 62 конфигурации).

На этом конфигурирование нашего кластера закончено. Осталось:

- скопировать файл конфигурации в директорию config бинарной сборки Ignite,
- скопировать наш "толстый" jar-файл в директорию libs пакета Ignite,
- перейти в консоли в корень пакета бинарной сборки Ignite и выполнить команду:

$ ./bin/ignite.sh -v ./config/ignite-config.xml

Apache Ignite cache: Запускаем сервис в Apache Ignite Service Grid. Часть 2

 Начнём с небольшой теории для тех, у кого не было времени ознакомиться с документацией Apache Ignite:

- нода - это один запущенный инстанс Apache Ignite. Документация рекомендует запускать одну ноду Ignite на одной JVM. Что, в грубом приближении, означает, что на одном сервере должна быть запущена только одна нода Apache Ignite. При этом в самой документации сказано, что есть способ запустить несколько нод Ignite на одной JVM, но использовать этот трюк для продакшена не рекомендуется, т.к. теряются все плюсы распределённого кеша : упал сервер - пропали сразу все запущенные на нём ноды со всеми своими кусочками кеша и его резервными копиями;

- кластер - это набор нод, объединённых в одну сеть, выполняющих общую задачу и взаимодействующих друг с другом по общим для них правилам: распределяющих между собой нагрузку, резервные копии кеша, выбирающих друг из друга ноду-координатора (в этом случае остальные ноды будут ведомыми) и т.д.

Ранее я уже писал о том, что в структуре бинарной сборки Apache Ignite есть директория libs, которая содержит библиотеки классов, которые Ignite загружает на старте. Эта директория отмечена цифрой "3" на скриншоте.

Соответственно, для того, чтобы наш сервис был загружен Игнайтом при старте кластера, нам нужно поместить в директорию libs скомпилированные классы нашего сервиса и всех его зависимостей, перечисленных в файле pom.xml. 

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

Если хотя бы на одной из таких нод в директории libs не будет классов нашего сервиса той же версии, что на остальных нодах, кластер Игнайта не стартует и выдаст нам ошибку, говорящую о том, что ноды находятся в несогласованном состоянии.

Поместить наш сервис с его зависимостями в директорию libs мы можем двумя способами:

1) Способ, рекомендованный документацией Apache Ignite.

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

2) Простой способ, который выбрал я.

Собирать наш сервис в "толстый" jar-файл - файл, содержащий все необходимые ему зависимости. И, при необходимости, обновлять наш сервис целиком со всеми зависимостями.

Собрать наш сервис в "толстый" jar-файл можно несколькими способами. Я выбрал maven-shade-plugin, т.к. мне он показался проще и понятнее остальных:

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

Apache Ignite: Запускаем сервис в Apache Ignite Service Grid. Часть 1

В первой части хочу рассказать о том, как у меня не получалось запустить наш сервис в Service Grid. Во второй - о том, как всё получилось.

Первая попытка запуска закончилась следующим исключением:

Как видно, причина исключения в том, что объект Gson не является сериализуемым. Речь идёт вот об этом поле объекта StartService (строка 27): 

Документация Apache Ignite говорит о том, что в этом случае поле должно быть помечено ключевым словом transient.

Ок. Помечаем поле ключевым словом transient и при следующем запуске получаем новое исключение:

Всеми любимый NullPointerException. Выбрасывается в объекте FileParser при попытке сохранить десериализованный объект в кеш:

Обкладываемся логгерами и видим, что в 36 строке у нас не происходит десериализации объекта Review из прочитанной из файла строки.

А функция десериализации объекта у нас - это:

Опять Gson! ))

Ситуация исправилась заменой поля Gson gson на локальную переменную метода StartService#execute():

Итог получился таким:

Задача стартовала, файл был загружен и распарсен.

Во второй части статьи я расскажу о том, какие настройки были сделаны мной в файле ignite-config.xml для того, чтобы наш сервис был инициализирован и запустил свою задачу при запуске кластера Apache Ignite.

Apache Ignite cache: объединяем все компоненты

Для объединения компонентов в одно приложение, нам понадобится проделать большую предварительную работу и как следует отрефакорить наше приложение, т.к. до сих пор оно запускалось у нас только отдельными компонентами в юнит-тестах.

1) Сделаем класс ActivityRepository, реализующий интерфейс CacheRepository, который будет сохранять информацию об активности приложения. Ранее, такой класс мы создавали только в качестве заглушки в тестовых целях. Сейчас делаем его способным работать с кешем Apache Ignite:

В конструктор ActivityRepository мы будем передавать инстанс класса, реализующего интерфейс Ignite, используя метод которого (getOrCreateCache()), мы будем создавать новый или получать уже созданный объект, реализующий интерфейс IgniteCache.


2) Создадим класс ReviewRepository, который будет отвечать за сохранение в кеш и получение из кеша данных, распаренных из нашего датасета:

ReviewRepository будет дополнительно реализовывать интерфейс org.apache.ignite.services.Service, что позволит нам запускать наш сервис в окружении Apache Ignite Service Grid - что-то вроде контейнера для теплая и запуска приложений. Подробнее об этом можно почитать в документации на сайте Apache Ignite.
Для чего нам нужно запускать ReviewRepository в Apache Ignite Service Grid? Это позволит нам в дальнейшем подключаться к ReviewRepository из любого внешнего приложения через специальный прокси, предоставляемый Ignite. Как это делается, мы рассмотрим далее. Для нетерпеливых - есть официальная документация.

3) И, наконец, реализуем наш главный, объединяющий всех, компонент - StartService:

Он, также как ReviewRepository, реализует интерфейс org.apache.ignite.services.Service, но, в отличие от ReviewRepository, полноценно реализует все его методы:

- init() - этот методе вызывается контейнером Service Grid  на стадии инициализации нашего сервиса. Пока здесь оставим заглушку, но в дальнейшем, используем и этот метод;

- execute() - этот метод вызывается после того, как сервис будет полностью инициализирован. Именно здесь мы разместим основную логику нашего сервиса.

Как видно из листинга кода, сначала мы стандартным способом, через оператор new, создаём инстансы  ActivityRepository, FileManager Downloader, FileParser. Затем создаём инстанс ReviewRepository через прокси, предоставляемый Apache Ignite для объектов, реализующих интерфейс org.apache.ignite.services.Service.

На своём старте, Apache Ignite проверяет наличие в специально отведённой директории классов для загрузки, загружает их, создаёт инстансы классов, реализующих интерфейс Service,  и вызывает их методы init, execute, cancel.

Т.к. инстанс ReviewRepository уже был создан Apache Ignite, нам нет необходимости создавать его снова. Вместо этого, мы можем использовать предоставленный нам прокси для вызова методов ReviewRepository.

После инициализации компонентов, мы создаём объект worker, реализующий интерфейс Runnable, который будет по расписанию выполнять всю полезную работу. 

Для начала, worker создаёт рабочие директории. Если не может создать, то завершает работу. Если рабочие директории были созданы успешно, worker даёт другим компонентам команды загрузить файл, распарсить его и сохранить в кеш.

Первый раз мы запускаем worker на выполнение сразу после создания. Затем, создаём scheduledExecutor и задаём ему расписание запуска worker’а.

- cancel() - этот метод вызывается при остановке нашего сервиса. Для остановки сервиса необходимо либо передать в Service Grid специальную команду, либо остановить кластер Apache Ignite.

4) Пишем утилитный класс для работы с датой и временем:

5) Ну, и не забываем писать юнит-тесты.

Пока оставим всё как есть и поверим юнит-тестам. В следующий раз запустим всё это добро вместе с Apache Ignite и посмотрим, не ошиблись ли мы в чём-нибудь.

Apache Ignite cache: разворачиваем Apache Ignite локально

Теперь, нетерпеливые мальчики и девочки, настало время самого интересного: разворачиваем Apache Ignite на своём локальном ПК. Это нам нужно для того, чтобы двигаться дальше в деле создания нашего приложения.

Описанные далее инструкции будут работать для Linux и MacOS. Желающие запустить Ignite на Windows могут посмотреть в официальной документации, как это правильно сделать.

Начать, конечно же, стоит с ознакомления с документацией Apache Ignite. Именно там описаны все возможные способы поднятия Ignite. 
По неизвестной мне причине, авторы документации “топят” за конфигурирование Ignite в java коде - именно в этом ключе дано большинство примеров в документации.

Но мы-то с вами уже ранее договорились, что вся конфигурация у нас должна быть во внешнем файле, чтобы дать возможность инженерам DevOps с ней работать без необходимости лезть в код нашего хоть и скоромного, но от того не менее непонятного для людей, далёких от java, приложения.

Начинаем с загрузки и распаковки бинарного пакета.

Что интересного в этом пакете:
1 - скрипт для запуска Ignite
2 - директория с конфигами (Ignite, логгера)
3 - директория с библиотеками классов, которые подтягивает Ignite на старте
4 - рабочая директория Ignite, куда он складывает созданные в ходе работы файлы (например, логи).

Запускаем Ignite в консоли командой:
./ignite.sh
 
Если хотим видеть в консоли подробные логи Ignite, то  добавляем флаг ‘-v’. В результате кластер запустится verbose-режиме и весь лог можно будет видеть в консоли:
./ignite.sh -v

Если хотим запустить кластер с нашей конфигурацией, которую мы в скором времени напишем, то запускаем кластер командой:
./ignite.sh -v /path/to/our/ignite-config.xml

Для начала, в качестве нашего первого конфига, возьмём из директории пакета копию файла конфигурации: ./config/default-config.xml и переименуем её в ignite-config.xml.
Тогда, если в терминале мы будем находиться в корневой директории пакета Apache Ignite, команда запуска у нас будет такой:
./bin/ignite.sh -v /config/ignite-config.xml

Для удобства я положу этот файл в репозиторий проекта в директорию configs.

Apache Ignite cache: пишем компонент, управляющий директориями на диске

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

Но потом вспомнил, что у нас ровно один файл, чьё название никогда не изменяется, и решил сделать компонент, который будет отвечать только за создание рабочей директории для приложения.

Рабочая директория нужна приложению для загрузки в неё файла с датасетом. Если такая директория отсутствует и её невозможно создать, то смысла в работе остальных компонентов приложения (загрузчик фалов и парсер) не будет. 

Получился вот такой класс FileManager:

@Slf4j
public class FileManager {

    /**
     * Create directory if not exists.
     *
     * @param path {@link String} path to the directory.
     * @return {@code true} if directory was created successful or directory already exists, {@code false} if got
     * Exception.
     */
    public boolean createIfNotExists(String path) {
        try {
            File dir = new File(path);
            if (!dir.exists()) {
                log.info("Create directory {}.", dir);
                return dir.mkdirs();
            } else {
                log.info("Directory {} already exists.", dir.getPath());
                return true;
            }
        } catch (Exception e) {
            log.error("Exception when try check or create directory: {}", path);
            return false;
        }
    }
}

Юнит-тест подтверждает, что этот класс работает так, как нам нужно:

public class FileManagerTest {

    private final String path = Objects.requireNonNull(getClass().getClassLoader().getResource(".")).getPath()
        + "test-work-dir";

    @Test
    public void createIfNotExists_shouldCreateDirectoriesInWorkDirectory() {
        FileManager fileManager = new FileManager();
        boolean result = fileManager.createIfNotExists(path);

        assertTrue(result);

        File dir = Paths.get(path).toFile();

        assertTrue(dir.exists());
        assertTrue(dir.isDirectory());

        assertTrue(dir.delete());
    }

    @Test
    public void createIfNotExists_shouldCorrectWorkIfWorkDirectoryAlreadyExists() {
        FileManager fileManager = new FileManager();
        boolean result1 = fileManager.createIfNotExists(path);
        boolean result2 = fileManager.createIfNotExists(path);

        assertTrue(result1);
        assertTrue(result2);

        File dir = Paths.get(path).toFile();

        assertTrue(dir.exists());
        assertTrue(dir.isDirectory());

        assertTrue(dir.delete());
    }
}

Apache Ignite cache: пишем компонент регистрации активности приложения

Наше приложение будет по расписанию запускаться, например, каждый час, скачивать нужный нам файл и проверять: если файл изменился, то необходимо распарсить его заново. Если не изменился, то ничего с ним делать не нужно.

Компонент регистрации активности приложения как раз и будет сохранять в кеш Apache Ignite информацию о том, какие файлы уже были обработаны нашим приложением.

Ключом для кеша будет имя файла (включая путь к файлу), а значением - модель Activity с полями: путь к файлу и хэш-сумма файла.

@Data
@AllArgsConstructor
public class Activity {
    private String path;
    private String checkSum;
}

Хэш-сумма файла будет вычисляться в приватном методе парсера файлов.

    private String getChecksum(String path) {
        try {
            MessageDigest md = MessageDigest.getInstance("SHA-256");
            md.update(Files.readAllBytes(Paths.get(path)));
            byte[] digest = md.digest();
            return DatatypeConverter.printHexBinary(digest).toUpperCase();
        } catch (Exception e) {
            log.error("Error get checksum for the file '{}'", path, e);
        }
        return null;
    }

Обновим тест парсера файлов, чтобы проверить, что один и тот же файл парсится только один раз.

    @Test
    public void parseFile_shouldParseTheSameFileOnlyOnce() {
        parser.parse(path, line -> gson.fromJson(line, Review.class), reviewRepository::save);
        parser.parse(path, line -> gson.fromJson(line, Review.class), reviewRepository::save);

        verify(reviewRepository, times(26)).save(any(Review.class));
        verify(activityRepository, times(2)).findById(path);
        verify(activityRepository, times(1)).save(any(Activity.class));

        assertEquals(26, reviewRepository.count());
        assertEquals(1, activityRepository.count());
    }