În acest tutorial simplu, vom crea câteva microservicii pe Spring Boot și vom organiza interacțiunea dintre ele prin cadrul Axon.
Să presupunem că avem această sarcină.
Există o sursă de tranzacții pe piața de capital. Această sursă ne transmite tranzacțiile printr-un API Rest.
Trebuie să obținem aceste tranzacții, să le salvăm în baza de date și să facem un depozit in-memory convenabil.
Această depozitare trebuie să îndeplinească următoarele funcții:
- să returneze o listă de tranzacții;
- să returneze poziția completă, adică tabela „instrument” – „cantitate curentă de acțiuni”;
- să returneze poziția pentru un instrument dat.
Cum ne vom apropia de rezolvarea acestei sarcini?
Conform principiilor modei microserviciilor, trebuie să împărțim sarcina în microservicii componente:
- obținerea tranzacției prin Rest;
- salvarea tranzacției în baza de date;
- depozit in-memory pentru prezentarea datelor pe poziție.
Să facem în cadrul acestui tutorial primul și al treilea serviciu, lăsând al doilea pentru a doua parte (scrieți în comentarii dacă asta vă interesează).
Așadar, avem două microservicii.
Primul obține date din exterior.
Al doilea procesează aceste date și răspunde la cererile primite.
Desigur, dorim să obținem scalabilitate orizontală, actualizări fără întreruperi și alte beneficii ale microserviciilor.
Ce sarcină, destul de complexă, se află în fața noastră?
De fapt, sunt multe, dar acum să discutăm despre cum vor circula datele între aceste microservicii. Se poate face REST între ele, se poate folosi o coadă, se poate gândi la multe soluții cu avantajele și dezavantajele lor.
Să analizăm una dintre abordările posibile – interacțiunea asincronă prin cadru Axon.
Care sunt avantajele acestei soluții?
În primul rând, interacțiunea asincronă crește flexibilitatea (da, există și un dezavantaj, dar deocamdată vorbim doar despre avantaje).
În al doilea rând, obținem direct din cutie Event Sourcing și CQRS.
În al treilea rând, Axon oferă o infrastructură gata făcută, iar noi trebuie să ne concentrăm doar pe dezvoltarea logicii de afaceri.
Să începem.
Proiectul nostru va fi pe gradle. Vor fi trei module:
- common. modul cu structuri de date comune (nu ne place copiarea);
- tradeCreator. modul cu microserviciul pentru primirea tranzacțiilor prin Rest;
- tradeQueries. modul cu microserviciul pentru afișarea poziției.
Vom lua Spring Boot ca bază și vom conecta starterul Axon.
Axon funcționează excelent și fără Spring, dar le vom folosi împreună.
Aici trebuie să ne oprim și să spunem câteva cuvinte despre Axon.
Este un sistem client-server. Există un server - este o aplicație separată pe care o vom rula în Docker.
Și există clienți, care sunt integrați în microservicii.
Deci, se conturează o imagine. Mai întâi, se pornește serverul Axon (în Docker), apoi microserviciile noastre.
La pornire, microserviciile caută serverul și încep să interacționeze cu acesta. Interacțiunea poate fi împărțită, în linii mari, în două tipuri: tehnic și de afaceri.
Tehnic - este un schimb de mesaje "sunt viu" (aceste mesaje pot fi văzute în modul de logare debug).
De afaceri - este un schimb de mesaje, de genul „nouă tranzacție”.
O caracteristică importantă, după pornire, microserviciul poate întreba serverul Axon „ce s-a întâmplat” și serverul transmite microserviciului evenimentele acumulate. Astfel, microserviciul poate fi repornit relativ sigur fără pierdere de date.
Într-o astfel de schemă de schimb, putem lansa foarte ușor multe instanțe de microservicii,
chiar și pe gazde diferite.
Da, o instanță a serverului Axon nu este de încredere, dar deocamdată așa este.
Lucrăm în paradigmele Event Sourcing și CQRS. Asta înseamnă că trebuie să avem „comenzi”, „evenimente” și „interogări”.
Vom avea o singură comandă: „creați tranzacția”, un singur eveniment „tranzacția a fost creată” și trei interogări: „afișează toate tranzacțiile”, „afișează poziția”, „afișează poziția pe instrument”.
Schema de funcționare este următoarea:
- Microserviciul tradeCreator primește tranzacția prin REST.
- Microserviciul tradeCreator creează comanda „creează tranzacția” și o trimite către serverul Axon.
- Serverul Axon primește comanda și o redirecționează către destinatarul interesat, în cazul nostru microserviciul tradeCreator.
- Microserviciul tradeCreator primește comanda, formează evenimentul „tranzacția a fost creată” și îl trimite serverului Axon.
- Serverul Axon primește evenimentul și îl redirecționează abonaților interesați.
- În prezent, avem doar un singur destinatar interesat - este microserviciul tradeQueries.
- Microserviciul tradeQueries primește evenimentul și își actualizează datele interne.
(Este important că, în momentul formării evenimentului, microserviciul tradeQueries poate să nu fie disponibil, dar de îndată ce va porni, va primi imediat evenimentul).
Da, serverul axon se află în centrul comunicațiilor, toate mesajele trec prin el.
Să trecem la codare.
Pentru a nu aglomera postarea cu cod, mai jos voi prezenta doar fragmente, linkul către exemplul complet va fi mai jos.
Să începem cu modulul general common.
În el, părțile comune sunt evenimentul (class CreatedTradeEvent). Observați denumirea, prin esență, aceasta este denumirea comenzii care a generat acest eveniment, dar la timpul trecut. În trecut, deoarece întâi apare comanda care duce la crearea evenimentului.
Alte structuri comune includ clasele pentru descrierea poziției (class Position), tranzacției (class Trade) și partea tranzacției (enum Side), adică cumpărare sau vânzare.
Să trecem la modulul tradeCreator.
Acest modul are un interfață Rest (class TradeController) pentru a primi tranzacții.
Din tranzacția primită se formează comanda „creați tranzacția” și este trimisă către serverul 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());
}
Pentru a procesa comanda se folosește clasa class TradeAggregate.
Pentru ca Axon să o găsească, adăugăm annotation @Aggregate.
Metoda pentru procesarea comenzii arată astfel (cu scurtare):
@CommandHandler
public TradeAggregate(CreateTradeCommand command) {
log.info("command: {}", command);
var event = CreatedTradeEvent.builder()
.tradeId(command.tradeId())
....
.build();
AggregateLifecycle.apply(event);
}
Din comandă se formează un eveniment și este trimis la server.
Comanda se află în clasa CreateTradeCommand.
Acum să ne uităm la ultimul modul tradeQueries.
Selectările sunt descrise în pachetul queries.
În acest modul există, de asemenea, o interfață Rest
public class TradeController.
Pentru exemplu, să vedem procesarea cererii: „afișează toate tranzacțiile”.
@GetMapping("/trade/all")
public List findAllTrades() {
return queryGateway.query(new FindAllTradesQuery(),
ResponseTypes.multipleInstancesOf(Trade.class)).join();
}
Se creează o cerere pentru selecție și este trimisă la server.
Pentru a procesa cererea de selecție se folosește clasa TradesEventHandler.
În ea există o metodă, marcată cu annotation
@QueryHandler
public List handleFindCurrentPositionQuery(FindCurrentPositionQuery query)
Aceasta este responsabilă pentru selectarea datelor din stocarea in-memory.
Se naște întrebarea, cum se actualizează informația în această stocare.
Să începem cu faptul că aceasta este pur și simplu un set de ConcurrentHashMap, conceput pentru selecții specifice.
Pentru actualizarea acestora se aplică metoda:
@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);
}El primește evenimentul „comerț creat” și actualizează hărțile.
Acestea sunt principalele puncte ale dezvoltării microserviciilor.
Ce se poate spune despre dezavantajele Axon?
În primul rând, complică infrastructura, a apărut un punct de eșec – serverul Axon, toate comunicațiile trec prin el.
În al doilea rând, dezavantajul acestor sisteme distribuite devine foarte evident – incoerența temporală a datelor. În cazul nostru, între primirea unei noi tranzacții și actualizarea datelor pentru interogări poate trece un timp inacceptabil de mult.
Ce a rămas în umbră?
Nu s-a spus nimic despre Event Sourcing și CQRS, ce sunt acestea și de ce sunt necesare.
Fără explicarea acestor concepte, unele aspecte pot fi neclare.
Este posibil ca unele fragmente de cod să necesite, de asemenea, explicații.
Despre aceasta vom vorbi la 21 septembrie.
.
Sursa: habr.com
