🥇Микросервизи с комуникация през Axon

В това просто ръководство ще създадем няколко микросервиса на Spring Boot и ще организираме взаимодействието между тях чрез фреймворка Axon.

Да предположим, че имаме такава задача.

Имаме източник на сделки на фондовия пазар. Този източник ни предава сделки чрез Rest-интерфейс.

Трябва да получим тези сделки, да ги запазим в база данни и да направим удобно in-memory хранилище.

Това хранилище трябва да изпълнява следните функции:

да връща списък с трейдове;
да връща пълната позиция, т.е. таблицата "инструмент" — "текущо количество ценни книжа";
да връща позиция по зададен инструмент.

Как ще подходим към решаването на тази задача?

Според принципите на микросервизната архитектура, трябва да разделим задачата на отделни микросервизи:

получаване на сделки чрез Rest;
запазване на сделките в база данни;
in-memory хранилище за представяне на данни по позиция.

Нека в рамките на това ръководство направим първия и третия сервис, а втория да оставим за втората част (пишете в коментарите, ако това е интересно).

Така, имаме два микросервиза.

Първият получава данни извън.

Вторият обработва тези данни и отговаря на входящите заявки.

Разбира се, искаме да постигнем хоризонтално мащабиране, безпроблемно обновление и други предимства на микросервизите.

Каква, доста сложна, задача стои пред нас?

Всъщност, те са много, но сега нека поговорим как ще се движат данните между тези микросервизи. Може да се направи и Rest, може да се постави някаква опашка, много неща могат да се измислят с техните плюсове и минуси.

Нека разгледаме един от възможните подходи – асинхронно взаимодействие чрез фреймворка Axon..

Какви са предимствата на такова решение?

На първо място, асинхронното взаимодействие увеличава гъвкавостта (да, тук има и минус, но засега говорим само за плюсовете).

На второ място, получаваме съхранение на събития и CQRS.
На трето място, Axon предоставя готова инфраструктура, и ние трябва да се съсредоточим само върху разработката на бизнес логиката.

Да започваме.

Проектът ни ще бъде на gradle. В него ще има три модула:

common. модул с общи структури от данни (не обичаме да копираме и поставяме);
tradeCreator. модул с микросервис за приемане на сделки чрез Rest;
tradeQueries. модул с микросервис за визуализиране на позиция.

Ще вземем Spring Boot за основа и ще свържем стартер Axon.

Axon работи отлично и без Spring, но ние ще ги използваме заедно.

Тук е нужно да се спрем и да кажем няколко думи за Axon.

Това е клиент-сървърна система. Имаме сървър – това е отделно приложение, което ще стартираме в Docker.

И имаме клиенти, които се вграждат в микросервизи.
Получава се следната картина. Първо се стартира Axon-сървър (в Docker), след това нашите микросервизи.

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

Техническото е обмен на съобщения "аз съм жив" (такива съобщения могат да се видят в режим на логване debug).

Бизнесовото е обмен на съобщения като "нова сделка".

Важна характеристика е, че след стартиране микросервизът може да попита Axon-сървъра "какво се е случило" и сървърът предава на микросервиза натрупаните събития. По този начин микросервизът може относително безопасно да бъде перезапуснат без загуба на данни.
При такава схема на обмен можем много лесно да стартираме много екземпляри на микросервизите,
включително на различни хостове.

Да, един екземпляр на Axon-сървъра не е надежден, но засега е така.

Работим в парадигмите Event Sourcing и CQRS. Това означава, че би трябвало да имаме "команди", "събития" и "избори".

Ще имаме една команда: "създай сделка", едно събитие "сделка създадена" и три избори: "покажи всички сделки", "покажи позиция", "покажи позиция по инструмент".

Схемата на работа изглежда така:

Микросервизът tradeCreator приема сделка по Rest.
Микросервизът tradeCreator създава команда "създай сделка" и я изпраща на Axon-сървъра.
Axon-сървърът приема командата и я препраща на заинтересования получател, в нашия случай това е микросервизът tradeCreator.
Микросервизът tradeCreator получава командата, формира събитие "сделка създадена" и го изпраща на Axon-сървъра.
Axon-сървърът приема събитието и го препраща на заинтересованите абонати.
В момента имаме само един заинтересован получател – микросервизът tradeQueries.
Микросервизът tradeQueries получава събитието и актуализира вътрешните данни.

(Важно е, че в моменти на формиране на събитието микросервизът tradeQueries може да не е достъпен, но веднага след като стартира, ще получи събитието).

Да, axon-сървърът стои в центъра на комуникациите, всички съобщения минават през него.

Нека да преминем към кодирането.

За да не затрупваме поста с код, по-долу ще дам само фрагменти, линкът към целия пример ще бъде по-долу.

Нека започнем с общия модул common.

В него общите части – това е събитие (class CreatedTradeEvent). Обърнете внимание на именуването, всъщност, това е името на командата, която е породила това събитие, но в миналото време. В миналото, тъй като първо се появява командата, която води до създаването на събитието.

Други общи структури включват класове за описание на позицията (class Position), сделката (class Trade) и страната на сделката (enum Side), т.е. купуване или продаване.

Преминаваме към модула tradeCreator.

Този модул има Rest-интерфейс (class TradeController) за приемане на сделки.
От получената сделка се формулира команда „създай сделка“ и се изпраща на axon-сървъра.

    @PostMapping("/trade")
    public ResponseEntity create(@RequestBody Trade trade) {
        var createTradeCommand = CreateTradeCommand.builder()
                .tradeId(trade.getTradeId())
	...
                .build();
        var result = commandGateway.sendAndWait(createTradeCommand, 3, TimeUnit.SECONDS);
        return ResponseEntity.ok(result.get().toString());
    }

За обработка на командата се използва класът class TradeAggregate.
За да може Axon да го намери, поставяме анотацията @Aggregate.
Методът за обработка на командата изглежда така (със съкращение):

    @CommandHandler
    public TradeAggregate(CreateTradeCommand command) {
        log.info("command: {}", command);
        var event = CreatedTradeEvent.builder()
                .tradeId(command.tradeId())
		....
                .build();
        AggregateLifecycle.apply(event);
    }

От командата се създава събитие и се изпраща на сървъра.
Командата се намира в класа CreateTradeCommand.

Сега да видим последния модул tradeQueries.

Извлеченията се описват в пакета queries.
В този модул също има Rest-интерфейс.
public class TradeController.

За пример да видим обработката на заявка: „покажи всички сделки“.

    @GetMapping("/trade/all")
    public List findAllTrades() {
        return queryGateway.query(new FindAllTradesQuery(),
                ResponseTypes.multipleInstancesOf(Trade.class)).join();
    }

Създава се заявка за извлечение и се изпраща на сървъра.

За обработка на заявка за извлечение се използва класът TradesEventHandler.
В него има метод, отбелязан с анотацията.

   @QueryHandler
    public List handleFindCurrentPositionQuery(FindCurrentPositionQuery query)

Той отговаря за извлечението на данни от in-memory хранилището.

Възниква въпросът, как в това хранилище се актуализира информацията.

Нека започнем с това, че това е просто набор от ConcurrentHashMap, пригодени за конкретни извлечения.
За тяхното актуализиране се прилага методът:

    @EventHandler
    public void on(CreatedTradeEvent event) {
        log.info("event:{}", event);

        var trade = Trade.builder()
	...
                .build();
        trades.put(event.tradeId(), trade);
        position.merge(event.shortName(), event.size(),
                (oldValue, value) -> event.side() == Side.BUY ? oldValue + value : oldValue - value);
    }

Той обработва събитие "създадена сделка" и актуализира Map-овете.

Това са основните точки на разработката на микросервизи.

Какво можем да кажем за недостатъците на Axon?

Първо, това усложнява инфраструктурата, появява се точка на отказ - Axon сървър, всички комуникации минават през него.

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

Какво остава неизказано?

Съвсем нищо не е казано за Event Sourcing и CQRS, какво са те и за какво са нужни.
Без обяснение на тези понятия, някои моменти могат да останат неразбрани.

Възможно е отделни фрагменти от кода също да изискват пояснение.

За това ще говорим на открит уебинар 21 септември.

Пълен пример.

Източник: habr.com

Микросервизи с комуникация чрез Axon