Reaktivni sustavi u Javi
1. Uvod
U ovom uputstvu razumjet ćemo osnove stvaranja reaktivnih sustava u Javi pomoću Springa i drugih alata i okvira.
U tom ćemo procesu razgovarati o tome kako je reaktivno programiranje samo pokretač stvaranja reaktivnog sustava. To će nam pomoći da razumijemo opravdanje za stvaranje reaktivnih sustava i različite specifikacije, knjižnice i standarde koje je nadahnuo tijekom svog rada.
2. Što su reaktivni sustavi?
Tijekom posljednjih nekoliko desetljeća tehnološki krajolik zabilježio je nekoliko poremećaja koji su doveli do potpune transformacije u načinu na koji vidimo vrijednost u tehnologiji. Računalni svijet prije Interneta nikada nije mogao zamisliti načine i sredstva na koje će promijeniti naš današnji dan.
Dosegom Interneta do masa i neprekidnim iskustvom koje obećava, arhitekti aplikacija moraju biti na nogama kako bi udovoljili svojoj potražnji.
U osnovi to znači da nikada ne možemo dizajnirati aplikaciju na način na koji smo to činili ranije. A visokoodzivna aplikacija više nije luksuz već potreba.
I to je suočeno sa slučajnim neuspjesima i nepredvidivim opterećenjem. Potreba sata nije samo za postizanjem točnog rezultata već i za brzim postizanjem! Prilično je važno pokrenuti nevjerojatna korisnička iskustva koja obećavamo pružiti.
To je ono što stvara potrebu za arhitektonskim stilom koji nam može dati reaktivne sustave.
2.1. Reaktivni manifest
Još 2013. godine, tim programera pod vodstvom Jonasa Bonera okupio se kako bi definirao skup temeljnih principa u dokumentu poznatom kao reaktivni manifest. To je ono što je postavilo temelje arhitektonskom stilu za stvaranje reaktivnih sustava. Od tada je ovaj manifest privukao veliko zanimanje zajednice programera.
U osnovi, ovaj dokument propisuje recept za reaktivni sustav koji treba biti fleksibilan, labavo povezan i skalabilan. To takve sisteme čini jednostavnim za razvoj, tolerantnim na kvarove i što je najvažnije vrlo brzo reagirajući, potpora nevjerojatnom korisničkom iskustvu.
Pa što je ovaj tajni recept? Pa, to je jedva neka tajna! Manifest definira temeljne karakteristike ili principe reaktivnog sustava:
- Uzvratni: Reaktivni sustav trebao bi osigurati brzo i dosljedno vrijeme odziva, a time i dosljednu kvalitetu usluge
- Otporna: Reaktivni sustav treba ostati reagirajući u slučaju slučajnih kvarova replikacijom i izolacijom
- Elastičan: Takav sustav trebao bi reagirati na nepredvidljiva radna opterećenja kroz isplativu skalabilnost
- Potaknut porukom: Trebao bi se oslanjati na asinkronu poruku koja prolazi između komponenata sustava
Ova načela zvuče jednostavno i razumno, ali nisu uvijek jednostavnija za primjenu u složenoj poslovnoj arhitekturi. U ovom uputstvu razvit ćemo uzorak sustava na Javi imajući na umu ove principe!
3. Što je reaktivno programiranje?
Prije nego što nastavimo, važno je razumjeti razliku između reaktivnog programiranja i reaktivnog sustava. Oba ova pojma koristimo prilično često i lako pogrešno razumijemo jedan za drugog. Kao što smo vidjeli ranije, reaktivni sustavi rezultat su specifičnog arhitektonskog stila.
U kontrastu, reaktivno programiranje je programska paradigma gdje je fokus na razvoju asinkronih i neblokirajućih komponenata. Jezgra reaktivnog programiranja je tok podataka koji možemo promatrati i reagirati na njega, čak i primijeniti povratni pritisak. To dovodi do neblokirajućeg izvršavanja, a time i do veće skalabilnosti s manje niti izvršenja.
To ne znači da se reaktivni sustavi i reaktivno programiranje međusobno isključuju. Zapravo je reaktivno programiranje važan korak ka ostvarenju reaktivnog sustava, ali nije sve!
3.1. Reaktivni tokovi
Reactive Streams je inicijativa zajednice koja je započela još 2013. godine pružaju standard za asinkronu obradu struje s neblokirajućim povratnim tlakom. Cilj je bio definirati skup sučelja, metoda i protokola koji mogu opisati potrebne operacije i entitete.
Od tada se pojavilo nekoliko implementacija u više programskih jezika koje odgovaraju specifikaciji reaktivnih tokova. Tu spadaju Akka Streams, Ratpack i Vert.x da nabrojimo samo neke.
3.2. Reaktivne knjižnice za Javu
Jedan od početnih ciljeva iza reaktivnih tokova trebao je na kraju biti uključen kao službena Java standardna knjižnica. Kao rezultat, specifikacija reaktivnih tokova semantički je ekvivalent knjižnici Java Flow, uvedenoj u Javi 9.
Osim toga, postoji nekoliko popularnih izbora za implementaciju reaktivnog programiranja u Javi:
- Reaktivna proširenja: Poznata pod nazivom ReactiveX, pružaju API za asinkrono programiranje s vidljivim streamovima. Dostupni su za više programskih jezika i platformi, uključujući Javu, gdje je poznata kao RxJava
- Project Reactor: Ovo je još jedna reaktivna knjižnica, osnovana na temelju specifikacije reaktivnih tokova, usmjerena na izgradnju ne-aplikacija na JVM-u. Također se događa da je temelj reaktivnog sloja u proljetnom ekosustavu
4. Jednostavna aplikacija
U svrhu ovog vodiča razvit ćemo jednostavnu aplikaciju koja se temelji na arhitekturi mikro usluga s minimalnim sučeljem. Arhitektura aplikacije trebala bi imati dovoljno elemenata za stvaranje reaktivnog sustava.
Za našu ćemo aplikaciju usvojiti reaktivno programiranje od kraja do kraja i druge obrasce i alate za postizanje temeljnih karakteristika reaktivnog sustava.
4.1. Arhitektura
Započet ćemo definiranjem jednostavna arhitektura aplikacije koja ne mora nužno pokazivati karakteristike reaktivnih sustava. Od tada ćemo napraviti potrebne promjene da bismo postigli ove karakteristike jednu po jednu.
Dakle, prvo, krenimo definiranjem jednostavne arhitekture:
Ovo je prilično jednostavna arhitektura koja ima hrpu mikrousluga za olakšavanje komercijalne upotrebe gdje možemo izvršiti narudžbu. Također ima sučelje za korisničko iskustvo, a sva komunikacija odvija se kao REST preko HTTP-a. Štoviše, svaka mikrousluga upravlja svojim podacima u pojedinačnim bazama podataka, što je praksa poznata kao baza podataka po usluzi.
Nastavit ćemo i stvoriti ovu jednostavnu aplikaciju u sljedećim pododjeljcima. Ovo će biti naše osnova za razumijevanje zabluda ove arhitekture te načine i sredstva za usvajanje načela i praksi kako bismo to mogli transformirati u reaktivni sustav.
4.3. Mikroservis inventara
Popis mikroservisa bit će odgovoran za upravljanje popisom proizvoda i njihovim trenutnim zalihama. Omogućit će i promjenu zaliha tijekom obrade narudžbi. Za razvoj ove usluge koristit ćemo Spring Boot s MongoDB.
Počnimo s definiranjem kontrolera za izlaganje nekih krajnjih točaka:
@GetMapping javni popis getAllProducts () {return productService.getProducts (); } @PostMapping javna narudžba processOrder (@RequestBody Order order) {return productService.handleOrder (narudžba); } @DeleteMapping javna narudžba revertOrder (@RequestBody Order order) {return productService.revertOrder (narudžba); }i usluga koja obuhvaća našu poslovnu logiku:
@Transactional public Order handleOrder (Order order) {order.getLineItems () .forEach (l -> {Product> p = productRepository.findById (l.getProductId ()) .orElseThrow (() -> new RuntimeException ("Nije moguće pronaći proizvod: "+ l.getProductId ())); if (p.getStock ()> = l.getQuantity ()) {p.setStock (p.getStock () - l.getQuantity ()); productRepository.save ( p);} else {baciti novi RuntimeException ("Proizvod je na zalihi:" + l.getProductId ());}}); vratiti nalog.setOrderStatus (OrderStatus.SUCCESS); } @Transactional public Order revertOrder (Order order) {order.getLineItems () .forEach (l -> {Product p = productRepository.findById (l.getProductId ()) .orElseThrow (() -> new RuntimeException ("Nije moguće pronaći proizvod: "+ l.getProductId ())); p.setStock (p.getStock () + l.getQuantity ()); productRepository.save (p);}); vratiti nalog.setOrderStatus (OrderStatus.SUCCESS); }Imajte na umu da jesmo ustrajavanje entiteta u transakciji, koji osigurava da u slučaju iznimaka ne dođe do nedosljednih stanja.
Osim njih, morat ćemo definirati i entitete domene, sučelje spremišta i hrpu klasa konfiguracije neophodnih da bi sve ispravno radilo.
No budući da su ovo uglavnom uzorci, izbjeći ćemo njihovo prolazak i na njih se može uputiti u spremište GitHub iz posljednjeg odjeljka ovog članka.
4.4. Dostava mikroservisa
Ni transportna mikro usluga neće se bitno razlikovati. Ovo će biti odgovoran za provjeru može li se za narudžbu generirati pošiljka i stvorite ga ako je moguće.
Kao i prije, definirat ćemo kontroler koji će izložiti naše krajnje točke, zapravo samo jednu krajnju točku:
@PostMapping postupak javne narudžbe (@RequestBody Narudžba) {return shippingService.handleOrder (narudžba); }i usluga za uvrštavanje poslovne logike koja se odnosi na otpremu narudžbi:
javni nalog handleOrder (nalog za narudžbu) {LocalDate shippingDate = null; if (LocalTime.now (). isAfter (LocalTime.parse ("10:00")) && LocalTime.now (). isBefore (LocalTime.parse ("18:00"))) {shippingDate = LocalDate.now () .plusDays (1); } else {baciti novi RuntimeException ("Trenutno je vrijeme izvan ograničenja za postavljanje narudžbe."); } shipmentRepository.save (new Shipment () .setAddress (order.getShippingAddress ()) .setShippingDate (shippingDate)); vratiti nalog.setShippingDate (datum isporuke) .setOrderStatus (OrderStatus.SUCCESS); }Naša jednostavna usluga otpreme samo provjerava valjani vremenski okvir za naručivanje. Izbjeći ćemo raspravu o ostatku šifre uzorka kao i prije.
4.5. Naručite Microservice
Na kraju ćemo definirati narudžbu mikrousluge koja će biti odgovoran za stvaranje novog poretka, osim ostalih stvari. Zanimljivo je da će igrati i kao usluga orkestratora gdje će komunicirati sa službom za inventar i brodskom službom za narudžbu.
Definirajmo svoj kontroler sa potrebnim krajnjim točkama:
@PostMapping stvaranje javne narudžbe (@RequestBody Narudžba narudžbe) {Narudžba obrađenaOrder = orderService.createOrder (narudžba); if (OrderStatus.FAILURE.equals (processingOrder.getOrderStatus ())) {throw new RuntimeException ("Obrada naloga nije uspjela, pokušajte ponovo kasnije."); } povratak obrađenNalog; } @GetMapping javni popis getAll () {return orderService.getOrders (); }I, usluga za inkapsuliranje poslovne logike koja se odnosi na narudžbe:
javni nalog createOrder (nalog za narudžbu) {boolean success = true; Narudžba savedOrder = orderRepository.save (narudžba); Narudžba inventaraResponse = null; isprobajte {inventoryResponse = restTemplate.postForObject (inventoryServiceUrl, order, Order.class); } catch (Iznimka ex) {success = false; } Naruči shippingResponse = null; isprobajte {shippingResponse = restTemplate.postForObject (shippingServiceUrl, order, Order.class); } catch (Iznimka ex) {success = false; HttpEntity deleteRequest = novi HttpEntity (redoslijed); ResponseEntity deleteResponse = restTemplate.exchange (inventoryServiceUrl, HttpMethod.DELETE, deleteRequest, Order.class); } if (uspjeh) {savedOrder.setOrderStatus (OrderStatus.SUCCESS); savedOrder.setShippingDate (shippingResponse.getShippingDate ()); } else {savedOrder.setOrderStatus (OrderStatus.FAILURE); } vratiti orderRepository.save (savedOrder); } javni popis getOrders () {return orderRepository.findAll (); }Rukovanje narudžbama kod kojih organiziramo pozive na usluge skladišta i otpreme daleko je od idealnog. Distribuirano transakcije s više mikroservisa složena je tema za sebe i izvan dosega ovog vodiča.
Međutim, vidjet ćemo kasnije u ovom vodiču kako reaktivni sustav može u određenoj mjeri izbjeći potrebu za distribuiranim transakcijama.
Kao i prije, nećemo proći kroz ostatak šifre uzorka. Međutim, to se može uputiti u GitHub repo.
4.6. Prednji kraj
Dodajmo i korisničko sučelje kako bi rasprava bila potpuna. Korisničko sučelje temeljit će se na Angular-u i bit će jednostavna aplikacija na jednoj stranici.
Trebat ćemo stvorite jednostavnu komponentu u Angulu za obradu naloga za izradu i dohvaćanje. Od posebne je važnosti dio u kojem zovemo naš API za stvaranje narudžbe:
createOrder () {let headers = new HttpHeaders ({'Content-Type': 'application / json'}); neka options = {headers: headers} this.http.post ('// localhost: 8080 / api / orders', this.form.value, options) .subscribe ((response) => {this.response = response}, (pogreška) => {this.error = pogreška})}Gornji isječak koda očekuje da će se podaci o narudžbi hvatati u obliku i biti dostupni u okviru komponente. Angular nudi fantastičnu podršku za stvaranje jednostavnih do složenih obrazaca pomoću reaktivnih obrazaca i obrazaca na temelju predloška.
Također je važan dio u kojem dobivamo prethodno stvorene narudžbe:
getOrders () {this.previousOrders = this.http.get ('' // localhost: 8080 / api / orders '')}Napominjemo da je Angular HTTP modul asinkrone prirode i stoga vraća RxJS Uočljivs. Mi se možemo nositi s odgovorom po našem mišljenju prolazeći ih kroz asinkronsku cijev:
Dosadašnje narudžbe:
ID narudžbe: {{order.id}}, status narudžbe: {{order.orderStatus}}, poruka narudžbe: {{order.responseMessage}}
Naravno, Angular će za rad trebati predloške, stilove i konfiguracije, ali na njih se može uputiti u spremište GitHub. Napominjemo da smo ovdje sve spojili u jednu komponentu, što u idealnom slučaju nije nešto što bismo trebali raditi.
Ali, za ovaj tutorial te zabrinutosti nisu u opsegu.
4.7. Postavljanje aplikacije
Sad kad smo stvorili sve pojedinačne dijelove aplikacije, kako bismo ih trebali primijeniti? Pa, to uvijek možemo ručno. Ali trebali bismo biti oprezni da uskoro može postati zamorno.
Za ovaj ćemo vodič koristiti Docker Compose za izgraditi i implementirati našu aplikaciju na Docker Machine. To će zahtijevati da dodamo standardni Dockerfile u svaku uslugu i stvorimo Docker Compose datoteku za cijelu aplikaciju.
Da vidimo kako ovo docker-compose.yml datoteka izgleda:
verzija: '3' usluge: frontend: build: ./frontend ports: - "80:80" order-service: build: ./order-service ports: - "8080: 8080" inventory-service: build: ./inventory -usluge: - "8081: 8081" usluga otpreme: build: ./shipping-service ports: - "8082: 8082"Ovo je prilično standardna definicija usluga u Docker Composeu i ne zahtijeva posebnu pažnju.
4.8. Problemi s ovom arhitekturom
Sad kad imamo jednostavnu aplikaciju s više usluga koje međusobno komuniciraju, možemo razgovarati o problemima u ovoj arhitekturi. Postoji nešto na što ćemo se pokušati pozabaviti u sljedećim odjeljcima i na kraju doći do stanja u kojem bismo našu aplikaciju transformirali u reaktivni sustav!
Iako je ova aplikacija daleko od proizvodnog softvera i postoji nekoliko problema, mi ćemo to učiniti usredotočiti se na pitanja koja se odnose na motivacije za reaktivne sustave:
- Neuspjeh u usluzi inventara ili usluzi otpreme može imati kaskadni učinak
- Pozivi vanjskim sustavima i bazama podataka blokiraju se u svojoj prirodi
- Uvođenje ne može automatski podnijeti kvarove i fluktuirajuća opterećenja
5. Reaktivno programiranje
Često blokiranje poziva u bilo kojem programu rezultiraju kritičnim resursima koji samo čekaju da se stvari dogode. To uključuje pozive baze podataka, pozive web uslugama i pozive datotečnog sustava. Ako uspijemo osloboditi niti izvršenja iz ovog čekanja i pružimo mehanizam za vraćanje u krug kad rezultati postanu dostupni, to će rezultirati mnogo boljim iskorištavanjem resursa.
To je ono što usvajanje paradigme reaktivnog programiranja čini za nas. Iako je moguće prebaciti se na reaktivnu knjižnicu za mnoge od tih poziva, možda neće biti moguće za sve. Srećom, Spring nam mnogo olakšava upotrebu reaktivnog programiranja s MongoDB i REST API-ima:
Spring Data Mongo ima podršku za reaktivni pristup putem Java upravljačkog programa MongoDB Reactive Streams. Pruža ReactiveMongoTemplate i ReactiveMongoRepository, oba koja imaju opsežnu funkcionalnost mapiranja.
Spring WebFlux pruža mrežni okvir reaktivnog sloga za Spring, omogućavajući neblokirajući kôd i povratni tlak reaktivnih tokova. Reaktor koristi kao reaktivnu knjižnicu. Nadalje, pruža WebClient za izvođenje HTTP zahtjeva s povratnim tlakom reaktivnih tokova. Koristi Reactor Netty kao knjižnicu HTTP klijenta.
5.1. Usluga inventara
Počet ćemo s promjenom krajnjih točaka da emitiraju reaktivne izdavače:
@GetMapping public Flux getAllProducts () {return productService.getProducts (); }@PostMapping javni Mono processOrder (@RequestBody Order order) {return productService.handleOrder (order); } @DeleteMapping javni Mono revertOrder (@RequestBody Order order) {return productService.revertOrder (order); }Očito ćemo morati izvršiti potrebne promjene i na usluzi:
@Transactional public Mono handleOrder (Order order) {return Flux.fromIterable (order.getLineItems ()) .flatMap (l -> productRepository.findById (l.getProductId ())) .flatMap (p -> {int q = order. getLineItems (). stream () .filter (l -> l.getProductId (). jednako (p.getId ())) .findAny (). get () .getQuantity (); if (p.getStock ()> = q) {p.setStock (p.getStock () - q); return productRepository.save (p);} else {return Mono.error (novi RuntimeException ("Proizvod je na zalihi:" + p.getId ()) );}}). then (Mono.just (order.setOrderStatus ("USPJEH"))); } @Transactional public Mono revertOrder (Order order) {return Flux.fromIterable (order.getLineItems ()) .flatMap (l -> productRepository.findById (l.getProductId ())) .flatMap (p -> {int q = order .getLineItems (). stream () .filter (l -> l.getProductId (). jednako (p.getId ())) .findAny (). get () .getQuantity (); p.setStock (p.getStock ( ) + q); vratiti productRepository.save (p);}). then (Mono.just (order.setOrderStatus ("SUCCESS"))); }5.2. Usluga otpreme
Slično tome, promijenit ćemo krajnju točku naše usluge otpreme:
@PostMapping javni mono postupak (@RequestBody Narudžba narudžbe) {return shippingService.handleOrder (narudžba); }I, odgovarajuće promjene u usluzi kako bi se iskoristilo reaktivno programiranje:
public Mono handleOrder (Order order) {return Mono.just (order) .flatMap (o -> {LocalDate shippingDate = null; if (LocalTime.now (). isAfter (LocalTime.parse ("10:00")) && LocalTime .now (). isBefore (LocalTime.parse ("18:00"))) {shippingDate = LocalDate.now (). plusDays (1);} else {return Mono.error (new RuntimeException ("Trenutno vrijeme je isključeno ograničenja za narudžbu. "));} vratiti pošiljkuRepository.save (nova Pošiljka () .setAddress (order.getShippingAddress ()) .setShippingDate (datum isporuke));}) .map (s -> order.setShippingDate (s. getShippingDate ()) .setOrderStatus (OrderStatus.SUCCESS)); }5.3. Usluga narudžbe
Morat ćemo napraviti slične promjene u krajnjim točkama usluge narudžbe:
@PostMapping javno Mono create (@RequestBody Order order) {return orderService.createOrder (order) .flatMap (o -> {if (OrderStatus.FAILURE.equals (o.getOrderStatus ())) {return Mono.error (new RuntimeException ( "Obrada narudžbe nije uspjela, pokušajte ponovo kasnije." + O.getResponseMessage ()));} else {return Mono.just (o);}}); } @GetMapping public Flux getAll () {return orderService.getOrders (); }Promjene na usluzi bit će više uključene jer ćemo morati iskoristiti Proljeće WebClient pozivati se na kraj inventara i isporuke reaktivnih krajnjih točaka:
javni Mono createOrder (nalog za narudžbu) {return Mono.just (order) .flatMap (orderRepository :: save) .flatMap (o -> {return webClient.method (HttpMethod.POST) .uri (inventoryServiceUrl) .body (BodyInserters.fromValue (o)) .exchange ();}) .onErrorResume (err -> {return Mono.just (order.setOrderStatus (OrderStatus.FAILURE) .setResponseMessage (err.getMessage ()));}) .flatMap (o -> {if (! OrderStatus.FAILURE.equals (o.getOrderStatus ())) {return webClient.method (HttpMethod.POST) .uri (shippingServiceUrl) .body (BodyInserters.fromValue (o)) .exchange ();} else { return Mono.just (o);}}) .onErrorResume (err -> {return webClient.method (HttpMethod.POST) .uri (inventoryServiceUrl) .body (BodyInserters.fromValue (order)) .retrieve () .bodyToMono (Order .class) .map (o -> o.setOrderStatus (OrderStatus.FAILURE) .setResponseMessage (err.getMessage ()));}) .map (o -> {if (! OrderStatus.FAILURE.equals (o.getOrderStatus ( ))) {return order.setShippingDate (o.getShippingDate ()) .setOrderStatus (OrderStatus.SUCCESS);} else {return order.setOrderStatus (OrderStatus.FAILURE) .setResponseMessage (o.getResponseMessage ()); }}) .flatMap (orderRepository :: save); } javni Flux getOrders () {return orderRepository.findAll (); }Ovaj vrsta orkestracije s reaktivnim API-jevima nije laka vježba i često je podložna pogreškama te je teško otkloniti pogreške. Vidjet ćemo kako se to može pojednostaviti u sljedećem odjeljku.
5.4. Prednji kraj
Sad, kad su naši API-ji sposobni strujati događaje onako kako se događaju, sasvim je prirodno da bismo to mogli iskoristiti i na našem front-endu. Srećom, Angular podržava EventSource, sučelje za događaje poslane s poslužitelja.
Pogledajmo kako možemo povući i obraditi sve svoje prethodne narudžbe kao tok događaja:
getOrderStream () {return Observable.create ((promatrač) => {neka eventSource = novi EventSource ('// localhost: 8080 / api / orders') eventSource.onmessage = (event) => {let json = JSON.parse ( event.data) this.orders.push (json) this._zone.run (() => {observer.next (this.orders)})} eventSource.onerror = (error) => {if (eventSource.readyState = == 0) {eventSource.close () this._zone.run (() => {observer.complete ()})} else {this._zone.run (() => {observer.error ('Pogreška izvora izvora: '+ pogreška)})}}})}}6. Arhitektura vođena porukama
Prvi problem kojem ćemo se pozabaviti povezan je s komunikacijom između usluga. Sada, ove su komunikacije sinkrone, što predstavlja nekoliko problema. To uključuje kaskadne kvarove, složenu orkestraciju i distribuirane transakcije da nabrojimo neke.
Očiti način rješavanja ovog problema je učiniti te komunikacije asinkronima. A posrednik poruka za olakšavanje sve komunikacije od usluge do usluge može učiniti trik za nas. Koristit ćemo Kafku kao posrednika poruka i Spring for Kafka za proizvodnju i potrošnju poruka:
Jednu ćemo temu koristiti za izradu i upotrebu poruka narudžbi s različitim statusima narudžbi kako bi usluge reagirale.
Pogledajmo kako se svaka usluga treba promijeniti.
6.1. Usluga inventara
Počnimo s definiranjem proizvođača poruka za našu uslugu inventara:
@Autowired private KafkaTemplate kafkaTemplate; javna void sendMessage (narudžba narudžbe) {this.kafkaTemplate.send ("narudžbe", narudžba); }Dalje, morat ćemo definirati potrošača poruke za uslugu inventara kako bi reagirao na različite poruke na temu:
@KafkaListener (topics = "naloga", groupId = "inventory") javna praznina troši (narudžba narudžbe) baca IOException {if (OrderStatus.RESERVE_INVENTORY.equals (order.getOrderStatus ())) {productService.handleOrder (order) .doOnSuc o -> {orderProducer.sendMessage (order.setOrderStatus (OrderStatus.INVENTORY_SUCCESS));}) .doOnError (e -> {orderProducer.sendMessage (order.setOrderStatus (OrderStatus.INVENTORY_FAILURE) .setRes (RES)). }). pretplatite se (); } else if (OrderStatus.REVERT_INVENTORY.equals (order.getOrderStatus ())) {productService.revertOrder (order) .doOnSuccess (o -> {orderProducer.sendMessage (order.setOrderStatus (OrderStatus.INVENTORY_REVERT_CESSORY). e -> {orderProducer.sendMessage (order.setOrderStatus (OrderStatus.INVENTORY_REVERT_FAILURE) .setResponseMessage (e.getMessage ()));}). pretplatite se (); }}To također znači da sada možemo sigurno ispustiti neke suvišne krajnje točke s našeg kontrolera. Te su promjene dovoljne za postizanje asinkrone komunikacije u našoj aplikaciji.
6.2. Usluga otpreme
Promjene u usluzi otpreme relativno su slične onim što smo ranije radili s uslugom inventara. Proizvođač poruka je isti, a potrošač poruke specifičan je za logiku otpreme:
@KafkaListener (topics = "naloga", groupId = "shipping") javna praznina troši (narudžba narudžbe) baca IOException {if (OrderStatus.PREPARE_SHIPPING.equals (order.getOrderStatus ())) {shippingService.handleOrder (order) .doOnSuccess (order) .doOnSuccess. o -> {orderProducer.sendMessage (order.setOrderStatus (OrderStatus.SHIPPING_SUCCESS) .setShippingDate (o.getShippingDate ()));}) .doOnError (e -> {orderProducer.sendMessage (order.setOrderStatusURE. (e.getMessage ()));}). pretplatite se (); }}Sada možemo sigurno ispustiti sve krajnje točke u naš kontroler jer nam više nisu potrebne.
6.3. Usluga narudžbe
Promjene u usluzi narudžbi bit će malo više uključene jer smo ovdje ranije radili sve orkestracije.
Unatoč tome, proizvođač poruka ostaje nepromijenjen, a potrošač poruke uzima logiku specifičnu za uslugu narudžbe:
@KafkaListener (topics = "naloga", groupId = "naloga") javna praznina troši (narudžba narudžbe) baca IOException {if (OrderStatus.INITIATION_SUCCESS.equals (order.getOrderStatus ())) {orderRepository.findById (order.getId () ) .map (o -> {orderProducer.sendMessage (o.setOrderStatus (OrderStatus.RESERVE_INVENTORY))); return o.setOrderStatus (order.getOrderStatus ()) .setResponseMessage (order.getResponseMessage () )ap}; : spremi) .subscribe (); } else if ("INVENTAR-USPJEH" .equals (order.getOrderStatus ())) {orderRepository.findById (order.getId ()) .map (o -> {orderProducer.sendMessage (o.setOrderStatus (OrderStatus.PREPARE_SHPPPP) ; povratak o.setOrderStatus (order.getOrderStatus ()) .setResponseMessage (order.getResponseMessage ());}) .flatMap (orderRepository :: save) .subscribe (); } else if ("SHIPPING-FAILURE" .equals (order.getOrderStatus ())) {orderRepository.findById (order.getId ()) .map (o -> {orderProducer.sendMessage (o.setOrderStatus (OrderStatus.REVERT_INVENTORY) ; povratak o.setOrderStatus (order.getOrderStatus ()) .setResponseMessage (order.getResponseMessage ());}) .flatMap (orderRepository :: save) .subscribe (); } else {orderRepository.findById (order.getId ()) .map (o -> {return o.setOrderStatus (order.getOrderStatus ()) .setResponseMessage (order.getResponseMessage ());}) .flatMap (orderRepository :: save ) .subscribe (); }}The potrošač ovdje samo reagira na poruke narudžbe s različitim statusima narudžbe. To je ono što nam daje koreografiju između različitih usluga.
Na kraju, naša usluga naručivanja također će se morati promijeniti kako bi podržala ovu koreografiju:
javni Mono createOrder (nalog za narudžbu) {return Mono.just (order) .flatMap (orderRepository :: save) .map (o -> {orderProducer.sendMessage (o.setOrderStatus (OrderStatus.INITIATION_SUCCESS)); return o;}). onErrorResume (err -> {return Mono.just (order.setOrderStatus (OrderStatus.FAILURE) .setResponseMessage (err.getMessage ()));}) .flatMap (orderRepository :: save); }Imajte na umu da je ovo mnogo jednostavnije od usluge koju smo morali napisati s reaktivnim krajnjim točkama u posljednjem odjeljku. Asinkroni koreografija često rezultira daleko jednostavnijim kodom, iako dolazi po cijenu eventualne dosljednosti i složenog otklanjanja pogrešaka i praćenja. Kao što pretpostavljamo, naš front-end više neće odmah dobiti konačni status narudžbe.
7. Usluga orkestracije kontejnera
Posljednji dio slagalice koji želimo riješiti povezan je s raspoređivanjem.
Ono što želimo u aplikaciji je velika suvišnost i tendencija automatskog povećanja ili smanjenja, ovisno o potrebi.
Već smo postigli kontejnerizaciju usluga putem Dockera i upravljamo ovisnostima između njih putem Docker Compose. Iako su ovo sami po sebi fantastični alati, oni nam ne pomažu da postignemo ono što želimo.
Dakle, mi trebate uslugu orkestracije spremnika koja se može pobrinuti za suvišnost i skalabilnost u našoj aplikaciji. Iako postoji nekoliko mogućnosti, jedna od popularnih uključuje Kubernetes. Kubernetes nam pruža agnostički dobavljač u oblaku za postizanje visoko skalabilnih implementacija kontejneriziranih radnih opterećenja.
Kubernetes omotava spremnike poput Dockera u mahune, koje su najmanja jedinica primjene. Nadalje, Deployment možemo koristiti za deklarativno opisivanje željenog stanja.
Implementacija stvara ReplicaSets, koji je interno odgovoran za podizanje mahuna. Možemo opisati minimalni broj identičnih mahuna koje bi trebale raditi u bilo kojem trenutku. To osigurava suvišnost, a time i visoku dostupnost.
Pogledajmo kako možemo definirati Kubernetesovu implementaciju za naše aplikacije:
apiVersion: apps / v1 vrsta: Metapodaci implementacije: ime: specifikacije rasporeda-raspoređivanja: replike: 3 selektor: matchLabels: ime: predložak rasporeda inventara: metapodaci: oznake: ime: inventar-raspoređivanje spec: spremnici: - naziv: slika inventara: inventory-service-async: najnoviji priključci: - containerPort: 8081 --- apiVersion: apps / v1 vrsta: Metapodaci implementacije: ime: specifikacije isporuke-implementacije: replike: 3 selektor: matchLabels: ime: predložak za isporuku: metapodaci: oznake : ime: specifikacija za otpremanje-otpremanje: spremnici: - ime: slika za otpremu: otprema-usluga-async: najnoviji priključci: - kontejnerPort: 8082 --- apiVersion: apps / v1 vrsta: metapodaci za postavljanje: ime: specifikacije za raspoređivanje: replike : 3 selektor: matchLabels: name: predložak za raspoređivanje naloga: metapodaci: oznake: ime: nalog za razmještanje: spremnici: - naziv: slika narudžbe: order-service-async: najnoviji portovi: - containerPort: 8080Ovdje izjavljujemo da ćemo u bilo kojem trenutku održavati tri identične replike mahuna. Iako je ovo dobar način za dodavanje suvišnosti, on možda neće biti dovoljan za različita opterećenja. Kubernetes nudi još jedan resurs poznat kao Horizontal Pod Autoscaler koji može skalirajte broj mahuna u rasporedu na temelju promatranih mjernih podataka poput upotrebe CPU-a.
Imajte na umu da smo upravo pokrili aspekte skalabilnosti aplikacije hostirane na Kubernetesovom klasteru. To ne znači nužno da je temeljni klaster sam po sebi skalabilan. Stvaranje Kubernetes klastera visoke dostupnosti netrivijalni je zadatak i izvan opsega ovog vodiča.
8. Rezultirajući reaktivni sustav
Sad kad smo napravili nekoliko poboljšanja u našoj arhitekturi, možda je vrijeme da to procijenimo prema definiciji reaktivnog sustava. Zadržat ćemo procjenu na temelju četiri karakteristike reaktivnog sustava o kojima smo raspravljali ranije u vodiču:
- Uzvratni: Usvajanje paradigme reaktivnog programiranja trebalo bi nam pomoći da postignemo neblokiranje s kraja na kraj, a time i odgovarajuću aplikaciju
- Otporna: Kubernetesova primjena s ReplicaSetom željenog broja mahuna trebala bi pružiti otpornost na slučajne kvarove
- Elastičan: Klaster Kubernetes i resursi trebali bi nam pružiti potrebnu potporu kako bismo bili elastični u slučaju nepredvidivih opterećenja
- Potaknut porukom: Kad bi se sva komunikacija usluga do usluge odvijala asinkrono putem Kafkinog brokera, ovdje bi nam trebalo pomoći
Iako ovo izgleda prilično obećavajuće, još uvijek nije gotovo. Da budem iskren, potraga za istinski reaktivnim sustavom trebala bi biti kontinuirana vježba poboljšanja. Nikada ne možemo spriječiti sve što može propasti u vrlo složenoj infrastrukturi, gdje je naša aplikacija samo mali dio.
Reaktivni sustav tako hoće zahtijevati pouzdanost od svakog dijela koji čini cjelinu. Od fizičke mreže do infrastrukturnih usluga poput DNS-a, svi bi trebali pasti u red da nam pomognu u postizanju krajnjeg cilja.
Često nam možda neće biti moguće upravljati i pružiti potrebna jamstva za sve ove dijelove. A ovo je gdje upravljana infrastruktura u oblaku pomaže nam ublažiti bol. Možemo birati između mnoštva usluga kao što su IaaS (Infeastrure-as-a-a-Service), BaaS (Backend-as-a-Service) i PaaS (Platform-as-a-Service) za prenošenje odgovornosti na vanjske strane. To nam ostavlja odgovornost za prijavu što je više moguće.
9. Zaključak
U ovom smo tutorijalu prošli kroz osnove reaktivnih sustava i kako se oni uspoređuju s reaktivnim programiranjem. Izradili smo jednostavnu aplikaciju s više mikroservisa i istakli probleme koje namjeravamo riješiti reaktivnim sustavom.
Dalje, nastavili smo, uvodeći reaktivno programiranje, arhitekturu zasnovanu na porukama i uslugu orkestracije spremnika u arhitekturi kako bismo realizirali reaktivni sustav.
Na kraju smo razgovarali o dobivenoj arhitekturi i o tome kako ona ostaje putovanje prema reaktivnom sustavu! Ovaj nas vodič ne upoznaje sa svim alatima, okvirima ili uzorcima koji nam mogu pomoći u stvaranju reaktivnog sustava, ali nas uvodi u putovanje.
Kao i obično, izvorni kod za ovaj članak može se naći na GitHubu.
