streamalism.org

1. Prehľad

V tomto rýchlom článku si predstavíme reaktorovú zbernicu nastavením scenára v reálnom živote pre reaktívnu aplikáciu riadenú udalosťami.

2. Základy projektového reaktora

2.1. Prečo reaktor?

Moderné aplikácie musia zvládnuť obrovské množstvo súbežných požiadaviek a spracovať značné množstvo údajov. Na splnenie týchto požiadaviek už nestačí štandardný blokovací kód.

Reaktívnym dizajnovým vzorom je architektonický prístup založený na udalostiach pre asynchrónne spracovanie veľkého množstva súbežných požiadaviek na službu pochádzajúce od jedného alebo viacerých spracovateľov služieb.

Projektový reaktor je založený na tomto vzore a má jasný a ambiciózny cieľ budovania neblokujúcich reaktívnych aplikácií na JVM..

2.2. Príklady scenárov

Predtým, ako začneme, je tu niekoľko zaujímavých scenárov, kde by využitie reaktívneho architektonického štýlu malo zmysel, len aby sme získali predstavu, kde by sme ho mohli uplatniť:

• Oznamovacie služby pre veľkú online nákupnú platformu, ako je Amazon
• Obrovské služby spracovania transakcií pre bankový sektor
• Akcie obchodujúce s akciami, kde sa ceny akcií menia súčasne

3. Závislosti Maven

Začnime používať Project Reactor Bus pridaním nasledujúcej závislosti do našej pom.xml:

 io.projectreactor reaktor-bus 2.0.8. UVOĽNENIE 

Môžeme skontrolovať najnovšiu verziu servera autobus-reaktor v Maven Central.

4. Vytvorenie ukážkovej aplikácie

Aby sme lepšie pochopili výhody prístupu založeného na reaktoroch, pozrime sa na praktický príklad.

Zostavíme jednoduchú aplikáciu zodpovednú za zasielanie upozornení používateľom online nákupnej platformy. Napríklad ak používateľ zadá novú objednávku, aplikácia odošle potvrdenie objednávky prostredníctvom e-mailu alebo SMS.

Typická synchrónna implementácia by bola prirodzene obmedzená priepustnosťou e-mailovej alebo SMS služby. Dopravné špičky, napríklad sviatky, by preto boli všeobecne problematické.

Reaktívnym prístupom môžeme navrhnúť náš systém tak, aby bol flexibilnejší a aby sa lepšie prispôsobil zlyhaniam alebo časovým limitom, ktoré sa môžu vyskytnúť v externých systémoch, ako sú servery brán.

Pozrime sa na aplikáciu - počnúc tradičnejšími aspektmi a pokračujeme k reaktívnejším konštrukciám.

4.1. Jednoduché POJO

Najprv vytvoríme triedu POJO, ktorá bude reprezentovať údaje upozornení:

public class NotificationData {private long id; súkromné ​​meno reťazca; súkromný reťazcový e-mail; súkromný mobilný reťazec; // metódy getra a setra}

4.2. Servisná vrstva

Definujme teraz jednoduchú vrstvu služieb:

verejné rozhranie NotificationService {void initiateNotification (NotificationData notificationData) vyvolá InterruptedException; }

A implementácia simulujúca dlhodobú operáciu:

@ Verejná trieda @Service NotificationServiceimpl implementuje NotificationService {@Override public void initiateNotification (NotificationData notificationData) hodí InterruptedException {System.out.println ("služba oznámenia bola spustená pre" + "ID oznámenia:" + notificationData.getId ()); Závit. Spánok (5 000); System.out.println ("Notifikačná služba sa skončila pre" + "ID oznámenia:" + notificationData.getId ()); }}

Všimnite si, že na ilustráciu skutočného scenára odosielania správ prostredníctvom SMS alebo e-mailovej brány zámerne zavádzame päťsekundové oneskorenie v InitiateNotification metóda s Závit. Spánok (5 000).

Následne, keď vlákno zasiahne službu, bude blokované na päť sekúnd.

4.3. Spotrebiteľ

Poďme teraz skočiť do reaktívnejších aspektov našej aplikácie a implementovať spotrebiteľa - ktorého potom namapujeme na zbernicu udalostí reaktora:

Verejná trieda @Service NotificationConsumer implementuje spotrebiteľa {@Autowired private NotificationService notificationService; @Override public void akceptovať (Event notificationDataEvent) {NotificationData notificationData = notificationDataEvent.getData (); skúsiť {notificationService.initiateNotification (notificationData); } catch (InterruptedException e) {// ignore}}}

Ako vidíme, spotrebiteľ, ktorého sme vytvorili, implementuje Spotrebiteľ rozhranie. Hlavná logika spočíva v súhlasiť metóda.

Toto je podobný prístup, s akým sa môžeme stretnúť pri typickej implementácii Spring Listener.

4.4. Kontrolór

Teraz, keď už dokážeme udalosti využiť, poďme ich tiež vygenerovať.

Urobíme to v jednoduchom ovládači:

@Controller public class NotificationController {@Autowired private EventBus eventBus; @GetMapping ("/ startNotification / {param}") public void startNotification (@PathVariable Integer param) {for (int i = 0; i <param; i ++) {NotificationData data = new NotificationData (); data.setId (i); eventBus.notify ("notificationConsumer", Event.wrap (údaje)); System.out.println ("Oznámenie" + i + ": úloha oznámenia bola úspešne odoslaná"); }}}

Je to celkom samozrejmé - udalosti vysielame prostredníctvom EventBus tu.

Napríklad, ak klient zasiahne URL s hodnotou parametra desať, potom sa prostredníctvom zbernice udalostí odošle desať udalostí.

4.5. Konfigurácia Java

Poďme teraz všetko spojiť a vytvorme jednoduchú aplikáciu Spring Boot.

Najprv musíme nakonfigurovať EventBus a Životné prostredie fazuľa:

@ Konfigurácia verejná trieda Konfigurácia {@Bean verejné prostredie env () {návrat Environment.initializeIfEmpty (). AssignErrorJournal (); } @Bean public EventBus createEventBus (Environment env) {return EventBus.create (env, Environment.THREAD_POOL); }}

V našom prípade vytvárame inštanciu EventBus s predvoleným fondom vlákien dostupným v prostredí.

Prípadne môžeme použiť prispôsobené Dispečer inštancia:

EventBus evBus = EventBus.create (env, Environment.newDispatcher (REACTOR_CAPACITY, REACTOR_CONSUMERS_COUNT, DispatcherType.THREAD_POOL_EXECUTOR));

Teraz sme pripravení vytvoriť hlavný kód aplikácie:

importovať statický reactor.bus.selector.Selectors. $; @SpringBootApplication verejná trieda NotificationApplication implementuje CommandLineRunner {@Autowired private EventBus eventBus; @Autowired private NotificationConsumer notificationConsumer; @Override public void run (String ... args) vyvolá Exception {eventBus.on ($ ("notificationConsumer"), notificationConsumer); } public static void main (String [] args) {SpringApplication.run (NotificationApplication.class, args); }}

V našom bežať metóda registrujeme notificationConsumer sa má spustiť, keď sa notifikácia zhoduje s daným selektorom.

Všimnite si, ako používame statický import súboru $ atribút na vytvorenie a Selektor objekt.

5. Vyskúšajte aplikáciu

Poďme si teraz vytvoriť test, aby sme videli naše NotificationApplication v akcii:

@RunWith (SpringRunner.class) @SpringBootTest (webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) verejná trieda NotificationApplicationIntegrationTest {@LocalServerPort súkromný int port; @Test public void givenAppStarted_whenNotificationTasksSubmitted_thenProcessed () {RestTemplate restTemplate = nový RestTemplate (); restTemplate.getForObject ("// localhost:" + port + "/ startNotification / 10", String.class); }}

Ako vidíme, hneď po vykonaní žiadosti všetkých desať úlohy sa odosielajú okamžite bez vytvárania blokovania. Po odoslaní sa udalosti oznámení budú spracovávať paralelne.

Oznámenie 0: úspešne odoslaná notifikačná úloha Oznámenie 1: úspešne odoslaná notifikačná úloha Oznámenie 2: úspešne odoslaná notifikačná úloha Oznámenie 3: úspešne odoslaná notifikačná úloha Oznámenie 4: úspešne odoslaná notifikačná úloha Oznámenie 5: úspešne odoslaná notifikačná úloha 7: notifikačná úloha úspešne odoslaná Notifikácia 8: notifikačná úloha úspešne odoslaná Notifikácia 9: notifikačná úloha úspešne odoslaná : 0 Notifikačná služba sa skončila pre identifikačné ID: 1 Notifikačná služba sa skončila pre identifikačné ID: 0 Notifikačná služba sa začala pre identifikačné ID: 4 Notifikačná služba sa skončila pre identifikačné ID: 3 skončilo pre ID oznámenia: 2 Spustila sa notifikačná služba pre ID oznámenia: 6 Spustila sa notifikačná služba pre ID oznámenia: 5 Spustila sa notifikačná služba pre ID oznámenia: 7 Notifikačná služba skončila pre ID oznámenia: 4 Spustila sa notifikačná služba pre ID oznámenia: 8 Notifikačné ID: 6 Notifikačná služba sa skončila pre identifikačné ID: 5 Notifikačná služba sa spustila pre identifikačné ID: 9 Notifikačná služba sa skončila pre identifikačné ID: 7 Notifikačná služba sa skončila pre identifikačné ID: 8 Notifikačná služba sa skončila pre identifikačné ID: 9

Je dôležité mať na pamäti to v našom scenári nie je potrebné tieto udalosti spracovávať v konkrétnom poradí.

6. Záver

V tomto rýchlom návode vytvorili sme jednoduchú aplikáciu založenú na udalostiach. Tiež sme videli, ako začať písať reaktívnejší a neblokujúci kód.

Avšak tento scenár iba poškriabe povrch subjektu a predstavuje len dobrý základ pre začatie experimentovania s reaktívnou paradigmou.

Ako vždy, zdrojový kód je k dispozícii na GitHub.