streamalism.org

1. Pregled

U ovom ćemo se članku usredotočiti na upotrebu reaktivnih proširenja (Rx) u Javi za sastavljanje i trošenje sekvenci podataka.

Na prvi pogled API može izgledati slično Java 8 Streams, ali zapravo je puno fleksibilniji i tečniji što ga čini moćnom paradigmom programiranja.

Ako želite pročitati više o RxJavi, pogledajte ovaj spis.

2. Postavljanje

Da bismo koristili RxJavu u našem projektu Maven, morat ćemo dodati sljedeću ovisnost u naš pom.xml:

 io.reactivex rxjava $ {rx.java.version} 

Ili, za projekt Gradle:

sastaviti 'io.reactivex.rxjava: rxjava: x.y.z'

3. Funkcionalni reaktivni koncepti

Na jednoj strani, funkcionalno programiranje je postupak izgradnje softvera sastavljanjem čistih funkcija, izbjegavanjem zajedničkog stanja, promjenjivih podataka i nuspojava.

Na drugoj strani, reaktivno programiranje je asinkrona programska paradigma koja se bavi protocima podataka i širenjem promjena.

Zajedno, funkcionalno reaktivno programiranje tvori kombinaciju funkcionalnih i reaktivnih tehnika koje mogu predstavljati elegantan pristup programiranju vođenom događajima - s vrijednostima koje se mijenjaju tijekom vremena i gdje potrošač reagira na podatke čim dođu.

Ova tehnologija objedinjuje različite implementacije svojih temeljnih principa, neki su autori smislili dokument koji definira uobičajeni rječnik za opisivanje nove vrste aplikacija.

3.1. Reaktivni manifest

Reactive Manifesto mrežni je dokument koji iznosi visoke standarde za aplikacije u industriji razvoja softvera. Jednostavno rečeno, reaktivni sustavi su:

• Odzivni - sustavi bi trebali reagirati na vrijeme
• Message Driven - sustavi bi trebali koristiti asinkroni prijenos poruka između komponenata kako bi osigurali labavu spregu
• Elastični - sustavi bi trebali ostati osjetljivi pod velikim opterećenjem
• Otporni - sustavi bi trebali reagirati kada neke komponente otkažu

4. Vidljive

Dvije su ključne vrste koje trebate razumjeti u radu s njima Rx:

• Uočljiv predstavlja bilo koji objekt koji može dobiti podatke iz izvora podataka i čije stanje može biti od interesa na način da drugi objekti mogu registrirati interes
• An posmatrač je bilo koji objekt koji želi biti obaviješten kada se stanje drugog objekta promijeni

An posmatrač pretplaćuje se na Uočljiv slijed. Slijed šalje stavke na posmatrač jedan po jedan.

The posmatrač obrađuje svaki prije obrade sljedećeg. Ako mnogi događaji uđu asinkrono, moraju se spremiti u red ili ispustiti.

U Rx, an posmatrač nikada neće biti pozvan sa stavkom koja nije u redu ili pozvan prije nego što se povratni poziv vrati za prethodnu stavku.

4.1. Vrste Uočljiv

Postoje dvije vrste:

• Neblokiranje - podržano je asinkrono izvršenje i dopušteno mu je otkazivanje pretplate u bilo kojem trenutku u toku događaja. U ovom ćemo se članku uglavnom usredotočiti na ovu vrstu

• Blokiranje - svi naSljedeće pozivi promatrača bit će sinkroni i nije moguće otkazati pretplatu usred toka događaja. Uvijek možemo pretvoriti Uočljiv u a Blokiranje uočljivo, koristeći metodu toBlocking:

BlockingObservable blockingObservable = observable.toBlocking ();

4.2. Operateri

An operater je funkcija koja uzima jedan Osagledljiv (izvor) kao prvi argument i vraća drugi Uočljiv (odredište). Tada će za svaku stavku koju izvorni izvor emitira primijeniti funkciju na tu stavku, a zatim emitirati rezultat na odredištu Uočljiv.

Operateri mogu se povezati lancima kako bi se stvorili složeni tokovi podataka koji filtriraju događaj na temelju određenih kriterija. Na isti se može primijeniti više operatora uočljiv.

Nije teško doći u situaciju u kojoj an Uočljiv emitira stavke brže od operater ili posmatrač mogu ih konzumirati. Više o povratnom pritisku možete pročitati ovdje.

4.3. Stvorite Promatljivo

Osnovni operator samo proizvodi an Uočljiv koji emitira jednu generičku instancu prije dovršenja, String "Zdravo". Kada želimo izvući informacije iz Uočljiv, provodimo posmatrač sučelje, a zatim nazovite pretplatite se na željeni Uočljivo:

Observable observable = Observable.just ("Hello"); observable.subscribe (s -> rezultat = s); assertTrue (result.equals ("Zdravo"));

4.4. OnNext, OnError, i OnCompleted

Postoje tri metode na posmatrač sučelje koje želimo znati o:

1. NaSljedeće je pozvan na naš posmatrač svaki put kad se u prilogu objavi novi događaj Uočljiv. Ovo je metoda kojom ćemo izvršiti neke radnje na svakom događaju
2. OnCompleted naziva se kada slijed događaja povezanih s Uočljiv je potpun, što ukazuje da više ne bismo trebali očekivati naSljedeće poziva našeg promatrača
3. OnError poziva se kada se tijekom obrade izbaci neobrađena iznimka RxJava okvirni kod ili naš kod za rukovanje događajima

Povratna vrijednost za Vidljivepretplatite se metoda je a pretplatite se sučelje:

Niz [] slova = {"a", "b", "c", "d", "e", "f", "g"}; Observable observable = Observable.from (slova); observable.subscribe (i -> rezultat + = i, // OnNext Throwable :: printStackTrace, // OnError () -> rezultat + = "_Completed" // OnCompleted); assertTrue (result.equals ("abcdefg_Completed"));

5. Uočljive transformacije i uvjetni operatori

5.1. Karta

Map operater pretvara stavke koje emitira Uočljiv primjenom funkcije na svaku stavku.

Pretpostavimo da postoji deklarirani niz nizova koji sadrži neka slova iz abecede i želimo ih ispisati u glavnom načinu:

Observable.from (slova) .map (String :: toUpperCase) .subscribe (slovo -> rezultat + = slovo); assertTrue (result.equals ("ABCDEFG"));

Ravna karta može se koristiti za poravnanje Vidljive kad god završimo s ugniježđenim Vidljive.

Više detalja o razlici između karta i flatMap možete pronaći ovdje.

Pod pretpostavkom da imamo metodu koja vraća Uočljiv s popisa žica. Sada ćemo ispisivati ​​za svaki niz iz novog Uočljiv popis naslova na temelju čega Pretplatnik vidi:

Observable getTitle () {return Observable.from (titleList); } Observable.just ("book1", "book2") .flatMap (s -> getTitle ()) .subscribe (l -> rezultat + = l); assertTrue (result.equals ("titletitle"));

5.2. Skenirati

The operator skeniranja aprimjenjuje funkciju na svaku stavku koju emitira Uočljiv sekvencijalno i emitira svaku uzastopnu vrijednost.

Omogućuje nam prijenos stanja iz događaja u događaj:

Niz [] slova = {"a", "b", "c"}; Observable.from (slova) .scan (novi StringBuilder (), StringBuilder :: append) .subscribe (ukupno -> rezultat + = total.toString ()); assertTrue (result.equals ("aababc"));

5.3. GroupBy

Grupiraj prema Operator omogućuje nam klasificiranje događaja u inputu Uočljiv u izlazne kategorije.

Pretpostavimo da smo stvorili niz cijelih brojeva od 0 do 10, a zatim primijenimo grupirati prema to će ih podijeliti u kategorije čak i neparan:

Observable.from (numbers) .groupBy (i -> 0 == (i% 2)? "EVEN": "ODD") .subscribe (group -> group.subscribe ((number) -> {if (group.getKey () .toString (). jednako ("ČAK")) {ČAK [0] + = broj;} ostalo {ODD [0] + = broj;}})); assertTrue (ČAK [0] .equals ("0246810")); assertTrue (ODD [0] .equals ("13579"));

5.4. filtar

Operater filtar emitira samo one predmete iz uočljiv koji prolaze a predikat test.

Pa filtrirajmo u cjelobrojni niz za neparne brojeve:

Observable.from (numbers) .filter (i -> (i% 2 == 1)) .subscribe (i -> result + = i); assertTrue (result.equals ("13579"));

5.5. Uvjetni operateri

DefaultIfEmpty emitira stavku iz izvora Uočljiv, ili zadana stavka ako je izvor Uočljiv prazno je:

Observable.empty () .defaultIfEmpty ("Observable is empty") .subscribe (s -> result + = s); assertTrue (result.equals ("Observable is empty"));

Sljedeći kod emitira prvo slovo abecede '' jer niz slova nije prazno i ​​ovo je ono što sadrži na prvom mjestu:

Observable.from (slova) .defaultIfEmpty ("Observable is empty") .first () .subscribe (s -> result + = s); assertTrue (result.equals ("a"));

TakeWhile operator odbacuje predmete koje emitira Uočljiv nakon što navedeni uvjet postane netačan:

Observable.from (numbers) .takeWhile (i -> i sum [0] + = s); assertTrue (zbroj [0] == 10);

Naravno, postoji još drugih operatera koji bi mogli zadovoljiti naše potrebe Sadrže, SkipWhile, SkipUntil, TakeUntil, itd.

6. Spojive opazive

A SpojivoOpažljivo sliči na običnu Uočljiv, osim što ne započinje s emitiranjem predmeta kad je pretplaćen na njega, već tek kad Spojiti na njega se primjenjuje operator.

Na taj način možemo pričekati da se svi predviđeni promatrači pretplate na Uočljiv prije Uočljiv započinje emitiranje predmeta:

Niz [] rezultat = {""}; ConnectableObservable connectable = Observable.interval (200, TimeUnit.MILLISECONDS) .publish (); connectable.subscribe (i -> rezultat [0] + = i); assertFalse (rezultat [0] .equals ("01")); connectable.connect (); Navoj.spavanje (500); assertTrue (rezultat [0] .equals ("01"));

7. Samac

Singl je poput Uočljiv koji umjesto emitiranja niza vrijednosti emitira jednu vrijednost ili obavijest o pogrešci.

S ovim izvorom podataka za pretplatu možemo koristiti samo dvije metode:

• OnUspjeh vraća a Singl koja također poziva metodu koju specificiramo
• OnError također vraća a Singl koji pretplatnike odmah obavještava o pogrešci

Niz [] rezultat = {""}; Pojedinačni singl = Observable.just ("Hello") .toSingle () .doOnSuccess (i -> rezultat [0] + = i) .doOnError (pogreška -> {baciti novi RuntimeException (error.getMessage ());}); single.subscribe (); assertTrue (rezultat [0] .equals ("Pozdrav"));

8. Ispitanici

A Predmet je istovremeno dva elementa, a pretplatnik i an uočljiv. Kao pretplatnik, subjekt se može koristiti za objavljivanje događaja koji dolaze iz više od jednog promatranog.

A budući da je također uočljiv, događaji više pretplatnika mogu se nadoknaditi kao njegovi događaji svakome tko ih promatra.

U sljedećem ćemo primjeru pogledati kako će promatrači moći vidjeti događaje koji se događaju nakon što se pretplate:

Cijeli pretplatnik1 = 0; Cijeli pretplatnik2 = 0; Observer getFirstObserver () {return new Observer () {@Override public void onNext (Integer value) {pretplatnik1 + = vrijednost; } @Override public void onError (Dostupno e) {System.out.println ("error"); } @Override public void onCompleted () {System.out.println ("Pretplatnik1 dovršen"); }}; } Observer getSecondObserver () {return new Observer () {@Override public void onNext (Integer value) {pretplatnik2 + = vrijednost; } @Override public void onError (Dostupno e) {System.out.println ("error"); } @Override public void onCompleted () {System.out.println ("Pretplatnik2 dovršen"); }}; } PublishSubject subject = PublishSubject.create (); subject.subscribe (getFirstObserver ()); subject.onNext (1); subject.onNext (2); subject.onNext (3); subject.subscribe (getSecondObserver ()); subject.onNext (4); subject.onCompleted (); assertTrue (pretplatnik1 + pretplatnik2 == 14)

9. Upravljanje resursima

Koristeći operacija omogućuje nam pridruživanje resursa, poput veze s bazom podataka JDBC, mrežne veze ili otvorenih datoteka, našim vidljivim.

Ovdje u komentarima predstavljamo korake koje moramo učiniti da bismo postigli ovaj cilj, a također i primjer provedbe:

Niz [] rezultat = {""}; Observable values ​​= Observable.using (() -> "MyResource", r -> {return Observable.create (o -> {for (Character c: r.toCharArray ()) {o.onNext (c);} o. onCompleted ();});}, r -> System.out.println ("Odloženo:" + r)); values.subscribe (v -> rezultat [0] + = v, e -> rezultat [0] + = e); assertTrue (rezultat [0] .equals ("MyResource"));

10. Zaključak

U ovom smo članku govorili o tome kako koristiti RxJava knjižnicu i o tome kako istražiti njene najvažnije značajke.

Cjeloviti izvorni kod za projekt, uključujući sve ovdje korištene uzorke koda, možete pronaći na Githubu.