streamalism.org

1. Áttekintés

Ebben a cikkben arra összpontosítunk, hogy a reaktív kiterjesztéseket (Rx) használjuk a Java-ban az adatok sorozatának összeállításához és felhasználásához.

Ránézésre az API hasonló lehet a Java 8 Stream-ekhez, de valójában sokkal rugalmasabb és gördülékenyebb, ami erőteljes programozási paradigmává teszi.

Ha többet szeretne megtudni az RxJaváról, nézze meg ezt az írást.

2. Beállítás

Az RxJava használatához a Maven projektünkben hozzá kell adnunk a következő függőséget pom.xml:

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

Vagy egy Gradle projekt esetében:

fordítsd le az „io.reactivex.rxjava: rxjava: x.y.z”

3. Funkcionális reaktív fogalmak

Az egyik oldalon, funkcionális programozás a szoftver felépítésének folyamata tiszta funkciók összeállításával, a megosztott állapot, a változtatható adatok és a mellékhatások elkerülésével.

A másik oldalon, reaktív programozás aszinkron programozási paradigma, amely az adatfolyamokkal és a változás terjedésével foglalkozik.

Együtt, funkcionális reaktív programozás funkcionális és reaktív technikák kombinációját alkotja, amely elegáns megközelítést jelenthet az eseményvezérelt programozáshoz - olyan értékekkel, amelyek idővel változnak, és ahol a fogyasztó reagál az adatokra, amikor bejönnek.

Ez a technológia egyesíti alapelveinek különböző megvalósításait, egyes szerzők előálltak egy dokumentummal, amely meghatározza a közös szókincset az új típusú alkalmazások leírására.

3.1. Reaktív kiáltvány

A Reaktív Kiáltvány egy online dokumentum, amely magas színvonalat határoz meg a szoftverfejlesztő ipar alkalmazásai számára. A reaktív rendszerek egyszerűen:

• Reszponzív - a rendszereknek időben kell reagálniuk
• Üzenetvezérelt - a rendszereknek aszinkron üzenetátvitelt kell használniuk az alkatrészek között a laza összekapcsolás biztosítása érdekében
• Rugalmas - a rendszereknek nagy terhelés mellett reagálniuk kell
• Rugalmas - a rendszereknek reagálniuk kell, ha egyes alkatrészek meghibásodnak

4. Megfigyelhetők

Két kulcsfontosságú típus érthető meg a munka során Rx:

• Megfigyelhető minden olyan objektumot képvisel, amely adatokat szerezhet adatforrásból, és amelynek állapota érdekes lehet oly módon, hogy más objektumok regisztrálhatnak érdeklődést
• An megfigyelő bármely olyan objektum, amely értesítést szeretne kapni, ha egy másik objektum állapota megváltozik

An megfigyelő feliratkozik egy Megfigyelhető sorrend. A sorrend elemeket küld a megfigyelő egyenként.

A megfigyelő mindegyiket kezeli, mielőtt feldolgozza a következőt. Ha sok esemény aszinkron módon érkezik, azokat sorban kell tárolni, vagy el kell dobni.

Ban ben Rx, an megfigyelő soha nem hívják meg soron kívüli tétellel, vagy addig hívják, amíg vissza nem érkezett az előző tétel visszahívása.

4.1. Típusok Megfigyelhető

Kétféle típus létezik:

• Nem blokkoló - az aszinkron végrehajtás támogatott, és az eseményfolyam bármely pontján leiratkozhat róla. Ebben a cikkben főleg az ilyen típusúakra fogunk koncentrálni

• Letiltás - minden onNext a megfigyelő hívások szinkronosak lesznek, és nem lehet leiratkozni egy eseményfolyam közepén. Mindig konvertálhatunk egy Megfigyelhető ba be Megfigyelhető blokkolás, a módszer használatával blokkolás:

BlockingObservable blockingObservable = observable.toBlocking ();

4.2. Operátorok

An operátor olyan funkció, amely egyet vesz Obszolgáltatható (a forrás) első argumentumaként ad vissza egy másikat Megfigyelhető (az úticél). Ezután minden olyan elemre, amelyet a forrás megfigyel, kiad egy függvényt az elemre, majd az eredményt a célállomáson bocsátja ki Megfigyelhető.

Operátorok összekapcsolva összetett adatfolyamok hozhatók létre, amelyek bizonyos szempontok alapján szűrik az eseményt. Több operátor alkalmazható ugyanarra megfigyelhető.

Nem nehéz olyan helyzetbe kerülni, amelyben egy Megfigyelhető gyorsabban bocsát ki elemeket, mint egy operátor vagy megfigyelő fogyaszthatja őket. Az ellennyomásról itt olvashat bővebben.

4.3. Hozzon létre Megfigyelhető

Az alapkezelő éppen előállít egy Megfigyelhető amely egyetlen általános példányt bocsát ki a String befejezése előtt "Helló". Amikor információt akarunk szerezni egy Megfigyelhető, megvalósítunk egy megfigyelő felületen, majd hívjon feliratkozást a kívántra Megfigyelhető:

Megfigyelhető megfigyelhető = Megfigyelhető.igaz ("Hello"); megfigyelhető.feliratkozás (s -> eredmény = s); assertTrue (eredmény.egyenlő ("Hello"));

4.4. OnNext, OnError, és OnCompleted

Három módszer létezik a megfigyelő felület, amelyről tudni akarunk:

1. OnNext hívják a mi megfigyelő minden alkalommal, amikor a csatolt új eseményt tesznek közzé Megfigyelhető. Ez az a módszer, ahol minden eseményt végrehajtunk
2. OnCompleted akkor hívják, amikor az eseményhez kapcsolódó események sorrendje Megfigyelhető befejeződött, jelezve, hogy nem szabad többet várnunk onNext felhívja megfigyelőnket
3. OnError akkor hívják meg, amikor a kezeletlen kivétel dobja a RxJava keretrendszer vagy eseménykezelő kódunk

A. Visszatérési értéke MegfigyelhetőkIratkozz fel módszer a Iratkozz fel felület:

Karakterlánc [] betűk = {"a", "b", "c", "d", "e", "f", "g"}; Megfigyelhető megfigyelhető = Megfigyelhető.-tól (betűk); observable.subscribe (i -> eredmény + = i, // OnNext Throwable :: printStackTrace, // OnError () -> eredmény + = "_Completed" // OnCompleted); assertTrue (eredmény.egyenlő ("abcdefg_Completed"));

5. Megfigyelhető átalakulások és feltételes operátorok

5.1. Térkép

A map operátor átalakítja az an által kibocsátott elemeket Megfigyelhető függvényt alkalmazva az egyes elemekre.

Tegyük fel, hogy van egy deklarált karakterlánc-tömb, amely tartalmaz néhány betűt az ábécéből, és nagybetűs módban szeretnénk kinyomtatni őket:

Megfigyelhető.from (betűk) .map (String :: toUpperCase) .subscribe (betű -> eredmény + = betű); assertTrue (eredmény.egyenlő ("ABCDEFG"));

A flatMap lapításhoz használható Megfigyelhetők valahányszor beágyazunk Megfigyelhetők.

További részletek a különbségről térkép és flatMap itt található.

Feltételezve, hogy van egy módszerünk, amely visszaadja az Megfigyelhető húrok listájáról. Most minden karakterláncot újból nyomtatunk Megfigyelhető mi alapján a címek listája Előfizető lát:

Megfigyelhető getTitle () {return Megfigyelhető.from (titleList); } Megfigyelhető.just ("könyv1", "könyv2") .flatMap (s -> getTitle ()) .feliratkozás (l -> eredmény + = l); assertTrue (eredmény.egyenlő ("titletitle"));

5.2. Letapogatás

A letapogató operátor afüggvényt alkalmaz egy minden által kibocsátott elemre Megfigyelhető egymás után és minden egyes egymást követő értéket kibocsát.

Ez lehetővé teszi számunkra, hogy az állapotot eseményről eseményre tovább vigyük:

Karakterlánc [] betű = {"a", "b", "c"}; Megfigyelhető.from (letters) .scan (új StringBuilder (), StringBuilder :: append) .subscribe (összesen -> eredmény + = total.toString ()); assertTrue (eredmény.egyenlő ("aababc"));

5.3. Csoportosít

Csoportosít operátor lehetővé teszi számunkra az események besorolását a bemenetben Megfigyelhető kimeneti kategóriákba.

Tegyük fel, hogy létrehoztunk egy 0 és 10 közötti egész tömböt, majd alkalmazzuk csoportosít ami fel fogja osztani őket a kategóriákra még és páratlan:

Megfigyelhető.from (számok) .groupBy (i -> 0 == (i% 2)? "EVEN": "ODD") .subscribe (group -> group.subscribe ((szám) -> {if (group.getKey) () .toString (). egyenlő ("EVEN")) {EVEN [0] + = szám;} else {ODD [0] + = szám;}})); assertTrue (EVEN [0] .egyenlő ("0246810")); assertTrue (ODD [0] .egyenlő ("13579"));

5.4. Szűrő

Az irányító szűrő csak azokat az elemeket bocsátja ki egy megfigyelhető hogy átmegy a állítmány teszt.

Szűrjünk be egy egész tömböt a páratlan számokra:

Megfigyelhető.-tól (számok). Szűrő (i -> (i% 2 == 1)). Előfizetés (i -> eredmény + = i); assertTrue (eredmény.egyenlő ("13579"));

5.5. Feltételes operátorok

DefaultIfEmpty tételt bocsát ki a forrásból Megfigyelhető, vagy egy alapértelmezett elem, ha a forrás Megfigyelhető üres:

Megfigyelhető.empty () .defaultIfEmpty ("A megfigyelhető üres") .feliratkozás (s -> eredmény + = s); assertTrue (result.equals ("A megfigyelhető üres"));

A következő kód az ábécé első betűjét adja kia ' mert a tömb leveleket nem üres, és ezt tartalmazza az első pozíció:

Megfigyelhető.from (betűk) .defaultIfEmpty ("A megfigyelhető üres") .első () .feliratkozás (s -> eredmény + = s); assertTrue (eredmény.egyenlő ("a"));

TakeWhile az üzemeltető eldobja a Megfigyelhető miután egy megadott feltétel hamissá válik:

Megfigyelhető.-tól (számok). TakeWhile (i -> i összeg [0] + = s); assertTrue (összeg [0] == 10);

Természetesen több olyan szolgáltató is működik, amely képes kielégíteni az igényeinket Tartalmazza, SkipWhile, SkipUntil, TakeUntil, stb.

6. Csatlakoztatható megfigyelhető elemek

A ConnectableObservable hasonlít egy közönségesre Megfigyelhető, azzal a különbséggel, hogy nem akkor kezd el elemeket bocsátani, amikor feliratkozott rá, hanem csak akkor, amikor a csatlakozzon operátort alkalmazzák rá.

Ily módon megvárhatjuk, amíg minden tervezett megfigyelő feliratkozik a Megfigyelhető előtte Megfigyelhető megkezdi a tárgyak kibocsátását:

Karakterlánc [] eredmény = {""}; ConnectableObservable connectable = Megfigyelhető.interval (200, TimeUnit.MILLISECONDS) .publish (); csatlakoztatható.feliratkozás (i -> eredmény [0] + = i); assertFalse (eredmény [0] .egyenlő ("01")); connectable.connect (); Szál.alszik (500); assertTrue (eredmény [0] .egyenlő ("01"));

7. Egyedülálló

Egyetlen olyan, mint egy Megfigyelhető aki ahelyett, hogy értékeket adna ki, egy értéket vagy hibaértesítést ad ki.

Ezzel az adatforrással csak két módszert alkalmazhatunk a feliratkozásra:

• OnSuccess visszatér a Egyetlen amely az általunk megadott módszert is meghívja
• OnError visszaadja a Egyetlen hogy azonnal értesíti az előfizetőket egy hibáról

Karakterlánc [] eredmény = {""}; Single single = Megfigyelhető.just ("Hello") .toSingle () .doOnSuccess (i -> eredmény [0] + = i) .doOnError (hiba -> {dobjon új RuntimeException (error.getMessage ());}); single.subscribe (); assertTrue (eredmény [0] .egyenlő ("Hello"));

8. Tantárgyak

A Tantárgy egyszerre két elem, a előfizető és egy megfigyelhető. Előfizetőként egy tantárgy felhasználható több megfigyelhető esemény eseményeinek közzétételére.

Mivel ez szintén megfigyelhető, a több előfizető eseményei újból elküldhetők eseményként bárkinek, aki megfigyeli.

A következő példában megnézzük, hogy a megfigyelők miként láthatják a feliratkozásuk után bekövetkező eseményeket:

Egész előfizető1 = 0; Egész előfizető2 = 0; Observer getFirstObserver () {return new Observer () {@Orride public void onNext (Integer value) {előfizető1 + = érték; } @Orride public void onError (Throwable e) {System.out.println ("hiba"); } @Orride public void onCompleted () {System.out.println ("Elfizető1 befejezve"); }}; } Megfigyelő getSecondObserver () {return new Observer () {@Orride public void onNext (Integer value) {előfizető2 + = érték; } @Orride public void onError (Throwable e) {System.out.println ("hiba"); } @Orride public void onCompleted () {System.out.println ("Előfizető2 befejezve"); }}; } PublishSubject subject = PublishSubject.create (); subject.subscribe (getFirstObserver ()); tárgy.onKövetkező (1); tárgy.onKövetkező (2); tárgy.onKövetkező (3); subject.subscribe (getSecondObserver ()); tárgy.onKövetkező (4); subject.onCompleted (); assertTrue (előfizető1 + előfizető2 == 14)

9. Erőforrás-kezelés

Használata A művelet lehetővé teszi számunkra, hogy erőforrásokat, például JDBC adatbázis-kapcsolatot, hálózati kapcsolatot vagy fájlokat nyissunk a megfigyelhető részeinkhez.

Itt bemutatjuk kommentárokban azokat a lépéseket, amelyeket meg kell tennünk a cél elérése érdekében, és példát is mutatunk a megvalósításra:

Karakterlánc [] eredmény = {""}; Megfigyelhető értékek = Megfigyelhető.használva (() -> "Saját erőforrás", r -> {return Megfigyelhető.create (o -> {for (Karakter c: r.toCharArray ()) {o.onNext (c);} o. onCompleted ();});}, r -> System.out.println ("Megsemmisítve:" + r)); értékek.feliratkozás (v -> eredmény [0] + = v, e -> eredmény [0] + = e); assertTrue (eredmény [0] .egyenlő ("Saját erőforrás"));

10. Következtetés

Ebben a cikkben arról beszéltünk, hogy hogyan kell használni az RxJava könyvtárat, és hogyan kell feltárni annak legfontosabb jellemzőit.

A projekt teljes forráskódja, beleértve az összes itt használt kódmintát, megtalálható a Github oldalon.