streamalism.org

1. Oversikt

I denne raske artikkelen introduserer vi reaktorbussen ved å sette opp et virkelig scenario for en reaktiv, hendelsesdrevet applikasjon.

2. Grunnleggende om Project Reactor

2.1. Hvorfor reaktor?

Moderne applikasjoner trenger å håndtere et stort antall samtidige forespørsler og behandle en betydelig mengde data. Standard blokkeringskode er ikke lenger tilstrekkelig for å oppfylle disse kravene.

Det reaktive designmønsteret er et hendelsesbasert arkitektonisk tilnærming for asynkron håndtering av et stort antall samtidige tjenesteforespørsler kommer fra enkelt- eller flere servicebehandlere.

Prosjektreaktoren er basert på dette mønsteret og har et klart og ambisiøst mål om å bygge ikke-blokkerende, reaktive applikasjoner på JVM.

2.2. Eksempel på scenarier

Før vi setter i gang, er det noen interessante scenarier der å utnytte den reaktive arkitektoniske stilen vil være fornuftig, bare for å få en ide om hvor vi kan bruke den:

• Varslingstjenester for en stor online shoppingplattform som Amazon
• Store transaksjonstjenester for banksektoren
• Aksjer handelsbedrifter der aksjekursene endres samtidig

3. Maven-avhengigheter

La oss begynne å bruke Project Reactor Bus ved å legge til følgende avhengighet i vår pom.xml:

 io.projektreaktorreaktor-buss 2.0.8.RELEASE 

Vi kan sjekke den siste versjonen av reaktor-buss i Maven Central.

4. Bygg en demo-applikasjon

For å bedre forstå fordelene med den reaktorbaserte tilnærmingen, la oss se på et praktisk eksempel.

Vi bygger et enkelt program som er ansvarlig for å sende varsler til brukerne av en online shoppingplattform. For eksempel, hvis en bruker legger inn en ny ordre, sender appen en ordrebekreftelse via e-post eller SMS.

En typisk synkron implementering vil naturligvis være begrenset av e-post eller SMS-tjenestens gjennomstrømning. Derfor vil trafikkpigger, for eksempel helligdager, generelt være problematiske.

Med en reaktiv tilnærming kan vi designe systemet vårt for å være mer fleksibelt og tilpasse seg bedre til feil eller tidsavbrudd som kan oppstå i eksterne systemer, for eksempel gateway-servere.

La oss ta en titt på applikasjonen - begynner med de mer tradisjonelle aspektene og går videre til de mer reaktive konstruksjonene.

4.1. Enkel POJO

La oss først lage en POJO-klasse for å representere varslingsdataene:

offentlig klasse NotificationData {privat lang id; privat strengnavn; privat streng e-post; privat String mobil; // getter og setter metoder}

4.2. Servicelaget

La oss nå definere et enkelt tjenestelag:

offentlig grensesnitt NotificationService {void initiateNotification (NotificationData notificationData) kaster InterruptedException; }

Og implementeringen, som simulerer en langvarig operasjon:

@Service public class NotificationServiceimpl implementerer NotificationService {@Override public void initiateNotification (NotificationData notificationData) kaster InterruptedException {System.out.println ("Notification service started for" + "Notification ID:" + notificationData.getId ()); Tråd. Søvn (5000); System.out.println ("Varslingstjeneste avsluttet for" + "Varslings-ID:" + notificationData.getId ()); }}

Legg merke til at vi med vilje introduserer en fem sekunders forsinkelse i det virkelige scenariet med å sende meldinger via en SMS eller e-postportal. initiateNotification metode med Tråd. Søvn (5000).

Følgelig, når en tråd treffer tjenesten, vil den bli blokkert i fem sekunder.

4.3. Forbrukeren

La oss nå hoppe inn i de mer reaktive aspektene av applikasjonen vår og implementere en forbruker - som vi deretter kartlegger til reaktorhendelsesbussen:

@Service offentlig klasse NotificationConsumer implementerer forbruker {@Autowired private NotificationService notificationService; @ Override public void accept (Event notificationDataEvent) {NotificationData notificationData = notificationDataEvent.getData (); prøv {notificationService.initiateNotification (notificationData); } fange (InterruptedException e) {// ignorere}}}

Som vi kan se, implementerer forbrukeren vi opprettet Forbruker grensesnitt. Hovedlogikken ligger i aksepterer metode.

Dette er en lignende tilnærming vi kan møte i en typisk vårlytterimplementering.

4.4. Kontrolleren

Til slutt, nå som vi er i stand til å konsumere hendelsene, la oss også generere dem.

Vi skal gjøre det i en enkel kontroller:

@Controller offentlig klasse NotificationController {@Autowired private EventBus eventBus; @GetMapping ("/ startNotification / {param}") offentlig ugyldig startNotification (@PathVariable Integer param) {for (int i = 0; i <param; i ++) {NotificationData data = new NotificationData (); data.setId (i); eventBus.notify ("notificationConsumer", Event.wrap (data)); System.out.println ("Varsling" + i + ": varslingsoppgave sendt vellykket"); }}}

Dette er ganske selvforklarende - vi sender ut hendelser gjennom EventBus her.

For eksempel, hvis en klient treffer URL-en med en param-verdi på ti, vil ti hendelser bli sendt gjennom hendelsesbussen.

4.5. Java Config

La oss nå sette alt sammen og lage en enkel Spring Boot-applikasjon.

Først må vi konfigurere EventBus og Miljø bønner:

@Configuration public class Config {@Bean public Environment env () {return Environment.initializeIfEmpty (). AssignErrorJournal (); } @Bean public EventBus createEventBus (Environment env) {return EventBus.create (env, Environment.THREAD_POOL); }}

I vårt tilfelle, vi instanserer EventBus med et standard trådbasseng tilgjengelig i miljøet.

Alternativt kan vi bruke en tilpasset Avsender forekomst:

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

Nå er vi klare til å lage en hovedprogramkode:

importer statisk reaktor.bus.selector.Selectors. $; @SpringBootApplication offentlig klasse NotificationApplication implementerer CommandLineRunner {@Autowired private EventBus eventBus; @Autowired private NotificationConsumer notificationConsumer; @Override public void run (String ... args) kaster unntak {eventBus.on ($ ("notificationConsumer"), notificationConsumer); } public static void main (String [] args) {SpringApplication.run (NotificationApplication.class, args); }}

I vår løpe metode vi registrerer varselForbruker som skal utløses når varselet samsvarer med en gitt velger.

Legg merke til hvordan vi bruker den statiske importen av $ attributt for å lage en Velger gjenstand.

5. Test applikasjonen

La oss nå lage en test for å se vår Varsel Søknad i aksjon:

@RunWith (SpringRunner.class) @SpringBootTest (webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) public class NotificationApplicationIntegrationTest {@LocalServerPort private int port; @Test offentlig ugyldighet gittAppStarted_whenNotificationTasksSubmitted_thenProcessed () {RestTemplate restTemplate = ny RestTemplate (); restTemplate.getForObject ("// localhost:" + port + "/ startNotification / 10", String.class); }}

Som vi kan se, så snart forespørselen er utført, alle ti oppgaver blir sendt umiddelbart uten å opprette noen blokkering. Og når de er sendt, blir varslingshendelsene behandlet parallelt.

Melding 0: varslingsoppgave ble sendt vellykket Melding 1: varslingsoppgave ble sendt vellykket Melding 2: varslingsoppgave ble sendt vellykket Melding 3: varslingsoppgave ble sendt vellykket Melding 4: varslingsoppgave ble sendt vellykket Melding 5: varslingsoppgave ble sendt vellykket Melding 6: varslingsoppgave sendt vellykket Melding 7: varslingsoppgave ble sendt vellykket Melding 8: varslingsoppgave sendt vellykket Melding 9: varslingsoppgave ble sendt vellykket Varslingstjeneste startet for varslings-ID: 1 Varslingstjeneste startet for varslings-ID: 2 Varslingstjeneste startet for varslings-ID: 3 Varslingstjeneste startet for varslings-ID : 0 Varslingstjeneste avsluttet for varslings-ID: 1 Varslingstjeneste avsluttet for varslings-ID: 0 Varslingstjeneste startet for varslings-ID: 4 Varslingstjeneste avsluttet for varslings-ID: 3 Varslingstjeneste avsluttet for Varslings-ID: 2 Varslingstjeneste startet for Varslings-ID: 6 Varslingstjeneste startet for Varslings-ID: 5 Varslingstjeneste startet for Varslings-ID: 7 Varslingstjeneste avsluttet for Varslings-ID: 4 Varslingstjeneste startet for Varslings-ID: 8 Varslingstjeneste avsluttet for Varslings-ID: 6 Varslingstjenesten avsluttet for Varslings-ID: 5 Varslingstjenesten startet for Varslings-ID: 9 Varslingstjenesten avsluttet for Varslings-ID: 7 Varslingstjenesten avsluttet for Varslings-ID: 8 Varslingstjenesten avsluttet for Varslings-ID: 9

Det er viktig å huske på det i vårt scenario er det ikke nødvendig å behandle disse hendelsene i en bestemt rekkefølge.

6. Konklusjon

I denne raske opplæringen, Vi har laget en enkel hendelsesdrevet applikasjon. Vi har også sett hvordan vi skal begynne å skrive en mer reaktiv og ikke-blokkerende kode.

Derimot, dette scenariet klør bare overflaten på motivet og representerer bare en god base for å begynne å eksperimentere med det reaktive paradigmet.

Som alltid er kildekoden tilgjengelig på GitHub.