Ebben az egyszerű útmutatóban elkészítünk néhány mikroszolgáltatást Spring Boot-on, és megszervezzük az interakciót közöttük az Axon keretrendszeren keresztül.
Tegyük fel, hogy van egy ilyen feladatunk.
Van egy forrásunk a tőzsdei tranzakciókhoz. Ez a forrás Rest interfészen keresztül küldi nekünk a tranzakciókat.
Ezeket a tranzakciókat meg kell kapnunk, el kell mentenünk az adatbázisba, és kényelmes in-memory tárolót kell készítenünk.
Ennek a tárolónak a következő funkciókat kell ellátnia:
- visszaadni a tranzakciók listáját;
- visszaadni a teljes pozíciót, azaz a «eszköz» – «jelenlegi papírmennyiség» táblázatot;
- visszaadni a pozíciót egy adott eszközre vonatkozóan.
Hogyan közelítjük meg ennek a feladatnak a megoldását?
A mikroszolgáltatási divat szabályai szerint a feladatokat alkotórészekre kell bontanunk:
- a tranzakciók lekérése Rest-en;
- a tranzakciók mentése az adatbázisba;
- in-memory tároló az adatok pozícióban történő megjelenítésére.
Tegyük meg az első és a harmadik szolgáltatást ebben az útmutatóban, a másodikat pedig hagyjuk a második részre (írjatok kommentben, ha ez érdekes).
Tehát, két mikroszolgáltatásunk van.
Az első külső adatokat kap.
A második feldolgozza ezeket az adatokat, és válaszol a bejövő kérésekre.
Természetesen szeretnénk elérni a vízszintes skálázhatóságot, a folyamatos frissítést, és egyéb mikroszolgáltatások előnyeit.
Milyen, nem éppen egyszerű feladat áll előttünk?
Valójában sok van, de most beszéljünk arról, hogy hogyan fognak az adatok áramlani ezek között a mikroszolgáltatások között. Készíthetünk Rest-et, beállíthatunk valamilyen sorban állást, sok mindenre gondolhatunk a saját előnyeinkkel és hátrányainkkal.
Nézzünk meg egy lehetséges megközelítést – asszinkron interakció az Axon keretrendszeren.
Mik az ilyen megoldás előnyei?
Először is, az asszinkron interakció növeli a rugalmasságot (igen, van itt egy hátrány, de most csak az előnyökről beszélünk).
Másodszor, közvetlenül a dobozból kapunk Eseményforrás és CQRS.
Harmadszor, az Axon kész infrastruktúrát biztosít, és nekünk csak a vállalati logika fejlesztésére kell összpontosítanunk.
Kezdjük.
A projektünket gradle-re alakítjuk. Három modult tartalmaz:
- common. modul a közös adatstruktúrákkal (nem szeretjük a másolást);
- tradeCreator. modul a tranzakciók Rest-en való fogadásához szükséges mikroszolgáltatással;
- tradeQueries. modul a pozíciók megjelenítésére szolgáló mikroszolgáltatással.
A Spring Boot-ot vesszük alapul és csatlakoztatjuk az Axon indítót.
Az Axon remekül működik Spring nélkül is, de együtt fogjuk őket használni.
Itt érdemes megállni, és pár szót mondani az Axonról.
Ez egy kliens-szerver rendszer. Van egy szerver – ez külön alkalmazás, amelyet Dockerben fogunk futtatni.
A kliensek pedig beépülnek a mikroszolgáltatásokba.
Így néz ki a helyzet. Először az Axon szervert indítjuk (Dockerben), majd a mikroszolgáltatásainkat.
Indításkor a mikroszolgáltatások keresik a szervert és kezdenek vele kapcsolatba lépni. A kapcsolatokat körülbelül két típusra bontjuk: technikai és üzleti.
A technikai – ez az üzenetek cseréje 'élek' (az ilyen üzeneteket láthatunk debug naplózási módban).
Az üzleti – ez az üzenetek cseréje, mint 'új üzlet'.
Fontos jellemző, hogy a mikroszolgáltatás a létrehozás után megkérdezheti az Axon szervertől, hogy 'mi történt', és a szerver átadja a mikroszolgáltatásnak a felhalmozott eseményeket. Így a mikroszolgáltatás viszonylag biztonságosan újraindítható adatvesztés nélkül.
Ezen az adatcsere sémán keresztül nagyon egyszerűen indíthatunk sok mikroszolgáltatás példányt,
ráadásul különböző hosztokon.
Igen, egy Axon szerver példány megbízhatatlan, de egyelőre így van.
Az Event Sourcing és a CQRS paradigmákban dolgozunk. Ez azt jelenti, hogy 'parancsaink', 'eseményeink' és 'lekérdezéseink' kell legyenek.
Lesz egy parancsunk: 'üzlet létrehozása', egy esemény 'üzlet létrejött', és három lekérdezés: 'minden üzlet megjelenítése', 'pozíció megjelenítése', 'pozíció megjelenítése eszközönként'.
A munkamenet ilyen lesz:
- A tradeCreator mikroszolgáltatás REST-en keresztül fogadja az ügyletet.
- A tradeCreator mikroszolgáltatás létrehoz egy 'üzlet létrehozása' parancsot, és elküldi azt az Axon szervernek.
- Az Axon szerver fogadja a parancsot és továbbítja azt az érdekelt félnek, esetünkben a tradeCreator mikroszolgáltatásnak.
- A tradeCreator mikroszolgáltatás megkapja a parancsot, létrehozza az 'üzlet létrejött' eseményt, és elküldi az Axon szervernek.
- Az Axon szerver fogadja az eseményt, és továbbítja az érdekelt feliratkozóknak.
- Jelenleg csak egy érdekelt fél van – a tradeQueries mikroszolgáltatás.
- A tradeQueries mikroszolgáltatás megkapja az eseményt és frissíti a belső adatokat.
(Fontos, hogy az esemény létrehozásának pillanatában a tradeQueries mikroszolgáltatás nem biztos, hogy elérhető, de amint elindul, azonnal megkapja az eseményt).
Igen, az Axon szerver áll a kommunikáció középpontjában, minden üzenet rajta keresztül megy.
Lépjünk a kódolásra.
Annak érdekében, hogy ne terheljük meg a hozzászólást kóddal, alább csak részleteket hozok, a teljes példára vonatkozó hivatkozás lentebb található.
Kezdjük a közös modulral, a common-nal.
Ebben a modulban találhatók a közös részek — ez az esemény (class CreatedTradeEvent). Figyeljen arra, hogy az elnevezés lényegében annak a csapatnak a neve, amely ezt az eseményt létrehozta, de múlt időben. Múlt időben, mivel először a csapat jelenik meg, amely az esemény létrehozásához vezet.
A többi közös struktúrához tartoznak a pozíció leírására szolgáló osztályok (class Position), a kereskedés (class Trade) és a kereskedési oldal (enum Side), azaz vásárlás vagy eladás.
Térjünk át a tradeCreator modulra.
Ez a modul Rest interfésszel rendelkezik (class TradeController) a kereskedések fogadására.
A kapott kereskedésből létrejön a "trade létrehozása" parancs, és elküldésre kerül az axon szerverre.
@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());
}
A parancs kezelésére a class TradeAggregate osztályt használjuk.
Ahhoz, hogy az Axon megtalálja, hozzá kell adnunk az @Aggregate annotációt.
A parancs kezelésére szolgáló metódus így néz ki (rövidítve):
@CommandHandler
public TradeAggregate(CreateTradeCommand command) {
log.info("parancs: {}", command);
var event = CreatedTradeEvent.builder()
.tradeId(command.tradeId())
....
.build();
AggregateLifecycle.apply(event);
}
A parancsból esemény keletkezik, és elküldésre kerül a szerverre.
A parancs a CreateTradeCommand osztályban található.
Most nézzük meg az utolsó tradeQueries modult.
A lekérdezések a queries csomagban találhatók.
Ebben a modulban is van Rest interfész.
public class TradeController.
Példaként nézzük meg a kérés kezelését: "minden kereskedés megjelenítése".
@GetMapping("/trade/all")
public List findAllTrades() {
return queryGateway.query(new FindAllTradesQuery(),
ResponseTypes.multipleInstancesOf(Trade.class)).join();
}
Egy lekérdezés készül, és elküldésre kerül a szerverre.
A lekérdezés kezelésére a TradesEventHandler osztályt használjuk.
Ebben van egy metódus, amelyet az annotációval jelölünk.
@QueryHandler
public List handleFindCurrentPositionQuery(FindCurrentPositionQuery query)
Ez felel a adatainak lekérdezéséért az in-memory tárolóból.
Felmerül a kérdés, hogyan frissül az információ ebben a tárolóban.
Kezdjük azzal, hogy ez egyszerűen a ConcurrentHashMap-ok halmaza, amelyeket bizonyos lekérdezésekhez optimalizáltak.
Ezek frissítésére a következő módszert alkalmazzuk:
@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);
}Ez fogadja a "trade létrejött" eseményt, és frissíti a Map-eket.
Ezek a mikroszolgáltatások fejlesztésének főbb pontjai.
Mit lehet mondani az Axon hátrányairól?
Először is, bonyolítja az infrastruktúrát, létrejön egy hibahely – az Axon szerver, minden kommunikáció ezen keresztül zajlik.
Másodszor, a hasonló elosztott rendszerek egyik legszembetűnőbb hátránya – az adatok időbeli inkonzisztenciája. Az esetünkben a új trade megérkezése és az adatok frissítése között túl sok idő telhet el.
Mi maradt elmondatlan?
Semmit nem mondtak az Event Sourcing-ról és CQRS-ról, hogy ezek mik és mire jók.
E fogalmak nélkül egyes pontok érthetetlenné válhatnak.
Lehet, hogy a kód egyes részei is magyarázatot igényelnek.
Erről fogunk beszélni a szeptember 21-én.
.
Forrás: habr.com
