Aszinkron programozás a Javához készült Azure SDK-ban

Ez a cikk a Javához készült Azure SDK aszinkron programozási modelljét ismerteti.

Az Azure SDK kezdetben csak nem blokkoló, aszinkron API-kat tartalmazott az Azure-szolgáltatásokkal való interakcióhoz. Ezek az API-k lehetővé teszik az Azure SDK használatát a rendszererőforrásokat hatékonyan használó skálázható alkalmazások létrehozásához. A JavaAzure SDK> azonban szinkronizált ügyfeleket is tartalmaz, hogy szélesebb közönséget kiszolgáljanak, és az ügyfélkódtárakat olyan felhasználók számára is elérhetővé tegyék, akik nem ismerik az aszinkron programozást. (Lásd: Megközelítés az Azure SDK tervezési irányelveiben.) Így az Azure SDK for Java összes Java-ügyfélkódtára aszinkron és szinkron ügyfeleket is kínál. Azonban éles rendszerekhez használjon aszinkron klienseket a rendszererőforrások maximális kihasználása érdekében.

Reaktív adatfolyamok az Azure SDK Java-hoz

Ha megtekinti a Async Szolgáltatásügyfél szakaszt a Java Azure SDK tervezési irányelvek, láthatja, hogy a Java 8 által biztosított CompletableFuture helyett az aszinkron API-k reaktív típusokat használnak. Miért választotta a Azure SDK csapat a JDK-ban natívan elérhető típusok közül a reaktív típusokat?

A Java 8 olyan funkciókat vezetett be, mint a Streams, a Lambdas és az CompletableFuture. Ezek a funkciók számos képességet biztosítanak, de bizonyos korlátozásokkal rendelkeznek.

CompletableFuture visszahívásalapú, nem blokkoló képességeket biztosít, és az CompletionStage interfész megkönnyíti az aszinkron műveletek sorozatának összeállítását. A Lambdas olvashatóbbá teszi ezeket a leküldéses alapú API-kat. A streamek funkcionális stílusú műveleteket biztosítanak az adatelemek gyűjteményének kezeléséhez. A streamek azonban szinkronok, és nem használhatók újra. CompletableFuture lehetővé teszi, hogy egyetlen kérést küldjön, támogatást nyújt a visszahíváshoz, és egyetlen választ vár. Számos felhőszolgáltatás azonban megköveteli az adatok streamelésének lehetőségét – például az Event Hubsot.

A reaktív streamek segíthetnek leküzdeni ezeket a korlátozásokat azáltal, hogy elemeket streamelnek egy forrásból egy előfizetőnek. Amikor egy előfizető adatokat kér egy forrástól, a forrás bármilyen számú eredményt küld vissza. Nem kell egyszerre elküldenie őket. Az átvitel egy adott időszakban történik, amikor a forrásnak van elküldendő adata.

Ebben a modellben az előfizető eseménykezelőket regisztrál az adatok beérkezésekor történő feldolgozásához. Ezek a leküldéses interakciók különböző jeleken keresztül értesítik az előfizetőt:

  • A onSubscribe() hívás azt jelzi, hogy az adatátvitel megkezdődött.
  • A onError() hívás azt jelzi, hogy hiba történt, ami az adatátvitel végét is jelzi.
  • A onComplete() hívás az adatátvitel sikeres befejezését jelzi.

A Java Streamekkel ellentétben a reaktív streamek első osztályú eseményként kezelik a hibákat. A reaktív streamek dedikált csatornával rendelkeznek a forrás számára, a hibáknak az előfizetővel való közléséhez. A reaktív adatfolyamok lehetővé teszik, hogy az előfizető tárgyalhassa az adatátviteli sebességet, és ezáltal a streameket egy push-pull modellé alakítsa át.

A Reaktív streamek specifikációja szabványt biztosít az adatok átvitelének módjához. Magas szinten a specifikáció a következő négy interfészt határozza meg, és meghatározza az interfészek implementálásának szabályait.

  • A Publisher egy adatfolyam forrása.
  • Az előfizető egy adatfolyam fogyasztója.
  • Az előfizetés kezeli a közzétevő és az előfizető közötti adatátvitel állapotát.
  • A processzor közzétevőként és előfizetőként egyaránt működik.

Néhány jól ismert Java kódtár biztosítja ennek a specifikációnak a megvalósítását, például RxJava, Akka Streams, Vert.x és Project Reactor.

A Java-hoz készült Azure SDK elfogadta a Project Reactort, hogy az aszinkron API-kat kínáljon. Ennek a döntésnek a fő mozgatórugója az volt, hogy zökkenőmentes integrációt biztosítson a Project Reactort is használó Spring Webfluxszal. A Project Reactor RxJava-n keresztül történő kiválasztásához az is hozzájárult, hogy a Project Reactor Java 8-at használ, de az RxJava akkor még a Java 7-nél volt. A Project Reactor számos olyan operátort is kínál, amelyek összeállíthatók, és lehetővé teszik deklaratív kód írását adatfeldolgozási folyamatok létrehozásához. A Project Reactor egy másik szép dolog, hogy adapterekkel rendelkezik a Project Reactor-típusok más népszerű implementálási típusokká való átalakításához.

Szinkron és aszinkron műveletek API-jainak összehasonlítása

Megismerkedett a szinkron ügyfelekkel és az aszinkron ügyfelek beállításaival. Az alábbi táblázat összefoglalja, hogyan néznek ki az API-k az alábbi lehetőségek használatával:

API-típus Nincs érték Egyetlen érték Több érték
Standard Java – Szinkron API-k void T Iterable<T>
Standard Java – Aszinkron API-k CompletableFuture<Void> CompletableFuture<T> CompletableFuture<List<T>>
Reaktív streamek felületei Publisher<Void> Publisher<T> Publisher<T>
Reaktív streamek projektreaktoros implementálása Mono<Void> Mono<T> Flux<T>

A teljesség kedvéért érdemes megemlíteni, hogy a Java 9 bevezette a Flow osztályt, amely tartalmazza a négy reaktív stream interfészt. Ez az osztály azonban nem tartalmaz implementációt.

Aszinkron API-k használata a Javához készült Azure SDK-ban

A reaktív streamek specifikációja nem tesz különbséget a közzétevők típusai között. A reaktív streamek specifikációjában a közzétevők egyszerűen nulla vagy több adatelemet állítanak elő. Sok esetben hasznos különbséget tenni kétféle közzétevő között: az egyik, amely legfeljebb egy adatelemet állít elő, és a másik, amely nullát vagy többet. A felhőalapú API-kban ez a különbség azt jelzi, hogy egy kérés egyetlen értékű választ vagy gyűjteményt ad vissza. A Project Reactor kétféleképpen teszi ezt a különbséget : Mono és Flux. Az az API, amely egy Mono választ ad vissza, legfeljebb egy értékkel rendelkezik, míg az az API, amely egy Flux választ ad vissza, nulla vagy több értéket tartalmaz.

Tegyük fel például, hogy egy ConfigurationAsyncClient használatával kéri le az Azure App Configuration service-ben tárolt konfigurációt. (További információ: Mi az Azure App Configuration?.)

Ha létrehoz egy ConfigurationAsyncClient az ügyfélen, és meghívja a getConfigurationSetting(), az egy Mono-et ad vissza, amely azt jelzi, hogy a válasz egyetlen értéket tartalmaz. A módszer önmagában való meghívása azonban nem tesz semmit. Az ügyfél még nem küldött kérelmet a Azure App Configuration szolgáltatásnak. Ebben a szakaszban az Mono<ConfigurationSetting> API által visszaadott adatok csak az adatfeldolgozási folyamat "szerelvényei". Ez az architektúra azt jelenti, hogy az adatok felhasználásához szükséges beállítás befejeződött. Az adatátvitel tényleges aktiválásához (azaz hogy a kérés eljusson a szolgáltatáshoz és megkapja a választ), elő kell fizetnie a visszaadott Mono-re. Tehát, amikor ezekkel a reaktív streamekkel foglalkozik, emlékeznie kell arra, hogy hívja meg subscribe(), mert semmi sem történik, amíg meg nem teszi.

Az alábbi példa bemutatja, hogyan iratkozhat fel a Mono konzolra, és hogyan nyomtathatja ki a konfigurációs értéket.

ConfigurationAsyncClient asyncClient = new ConfigurationClientBuilder()
    .connectionString("<your connection string>")
    .buildAsyncClient();

asyncClient.getConfigurationSetting("<your config key>", "<your config value>").subscribe(
    config -> System.out.println("Config value: " + config.getValue()),
    ex -> System.out.println("Error getting configuration: " + ex.getMessage()),
    () -> System.out.println("Successfully retrieved configuration setting"));

System.out.println("Done");

Figyelje meg, hogy az ügyfél meghívása getConfigurationSetting() után a példakód feliratkozik az eredményre, és három különálló lambdát biztosít. Az első lambda a szolgáltatástól kapott adatokat használja fel, amelyek sikeres válasz esetén aktiválódnak. A második lambda akkor aktiválódik, ha hiba történik a konfiguráció beolvasása közben. A harmadik lambda az adatfolyam befejezésekor lesz meghívva, ami azt jelenti, hogy nem várható több adatelem ebből a streamből.

Megjegyzés:

Az összes aszinkron programozáshoz hasonlóan az előfizetés létrehozása után a végrehajtás a szokásos módon folytatódik. Ha nincs semmi, amely a program aktív és végrehajtható maradna, az az aszinkron művelet befejeződése előtt leállhat. Az a fő szál, amely meghívta a subscribe() elemet, nem várakozik addig, amíg hálózati hívást kezdeményezel az Azure App Configurationhöz, és választ nem kapsz. Az éles rendszerekben előfordulhat, hogy valami mást is feldolgoz, de ebben a példában kis késést adhat hozzá, ha meghívja a Thread.sleep() függvényt, vagy használ egy CountDownLatch-et, hogy lehetőséget adjon az aszinkron művelet befejeződésére.

Az alábbi példában látható módon az olyan API-k, amelyek egy Flux típusú értéket adnak vissza, szintén hasonló mintát követnek. A különbség az, hogy a metódushoz subscribe() megadott első visszahívást a rendszer többször hívja meg a válasz minden adateleméhez. A hiba vagy a befejezési visszahívások pontosan egyszer lesznek meghívva, és termináljeleknek minősülnek. A rendszer nem hív meg más visszahívásokat, ha ezek közül bármelyik jel a közzétevőtől érkezik.

EventHubConsumerAsyncClient asyncClient = new EventHubClientBuilder()
    .connectionString("<your connection string>")
    .consumerGroup("<your consumer group>")
    .buildAsyncConsumerClient();

asyncClient.receive().subscribe(
    event -> System.out.println("Sequence number of received event: " + event.getData().getSequenceNumber()),
    ex -> System.out.println("Error receiving events: " + ex.getMessage()),
    () -> System.out.println("Successfully completed receiving all events"));

Visszanyomás

Mi történik, ha a forrás gyorsabban készíti el az adatokat, mint amennyit az előfizető képes kezelni? Az előfizetőt túlterhelhetik az adatok, ami memóriahibákhoz vezethet. Az előfizetőnek vissza kell kommunikálnia a közzétevővel, hogy lassítson, ha nem tud lépést tartani. Alapértelmezés szerint, amikor a korábbi példában látható módon meghívja a subscribe() függvényt egy Flux-n, az előfizető nem korlátozott adatfolyamot kér, ezzel jelezve a közlőnek, hogy a lehető leggyorsabban küldje az adatokat. Ez a viselkedés nem mindig kívánatos, és előfordulhat, hogy az előfizetőnek szabályoznia kell a közzététel sebességét a "backpressure" használatával. A backpressure lehetővé teszi az előfizető számára az adatelemek áramlásának szabályozását. Az előfizető korlátozott számú olyan adatelemet kér, amelyet kezelni tud. Miután az előfizető befejezte ezeknek az elemeknek a feldolgozását, további kéréseket kérhet. A backpressure használatával a leküldéses adatátviteli modellt leküldéses-lekéréses modellté alakíthatja.

Az alábbi példa bemutatja, hogyan szabályozhatja, hogy az Eseményközpontok fogyasztója milyen sebességgel fogad eseményeket:

EventHubConsumerAsyncClient asyncClient = new EventHubClientBuilder()
    .connectionString("<your connection string>")
    .consumerGroup("<your consumer group>")
    .buildAsyncConsumerClient();

asyncClient.receive().subscribe(new Subscriber<PartitionEvent>() {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1); // request 1 data element to begin with
    }

    @Override
    public void onNext(PartitionEvent partitionEvent) {
        System.out.println("Sequence number of received event: " + partitionEvent.getData().getSequenceNumber());
        this.subscription.request(1); // request another event when the subscriber is ready
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("Error receiving events: " + throwable.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("Successfully completed receiving all events")
    }
});

Amikor az előfizető először "csatlakozik" a közzétevőhöz, a közzétevő átad egy példányt Subscription az előfizetőnek, amely kezeli az adatátvitel állapotát. Ez Subscription az a közeg, amelyen keresztül az előfizető visszanyomást alkalmazhat, ha meghívja request() , hogy adja meg, hány további adatelemet képes kezelni.

Ha az előfizető minden híváskor onNext()több adatelemet kér, request(10) például a közzétevő azonnal elküldi a következő 10 elemet, ha elérhetők, vagy amikor elérhetővé válnak. Ezek az elemek az előfizető oldalán egy pufferben halmozódnak fel, és mivel minden onNext() hívás 10 további elemet kér, a torlódás addig növekszik, amíg либо a közzétevőnek nem marad több elküldhető adateleme, либо az előfizető puffere túl nem csordul, ami memóriahiány miatti hibákat eredményez.

Reaktív stream-előfizetés lemondása

Az előfizetés kezeli a közzétevő és az előfizető közötti adatátvitel állapotát. Az előfizetés addig marad aktív, amíg a közzétevő nem küldi át az összes adatot az előfizetőnek, vagy az előfizető már nem szeretne adatokat kapni. Az előfizetést több módon is lemondhatja, ahogyan az alábbi példákban is látható.

Az alábbi példa megszakítja az előfizetést az előfizető kizárásával:

EventHubConsumerAsyncClient asyncClient = new EventHubClientBuilder()
    .connectionString("<your connection string>")
    .consumerGroup("<your consumer group>")
    .buildAsyncConsumerClient();

Disposable disposable = asyncClient.receive().subscribe(
    partitionEvent -> {
        Long num = partitionEvent.getData().getSequenceNumber()
        System.out.println("Sequence number of received event: " + num);
    },
    ex -> System.out.println("Error receiving events: " + ex.getMessage()),
    () -> System.out.println("Successfully completed receiving all events"));

// much later on in your code, when you are ready to cancel the subscription,
// you can call the dispose method, as such:
disposable.dispose();

A következő példa az cancel() metódus meghívásával megszakítja az Subscription előfizetést.

EventHubConsumerAsyncClient asyncClient = new EventHubClientBuilder()
    .connectionString("<your connection string>")
    .consumerGroup("<your consumer group>")
    .buildAsyncConsumerClient();

asyncClient.receive().subscribe(new Subscriber<PartitionEvent>() {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1); // request 1 data element to begin with
    }

    @Override
    public void onNext(PartitionEvent partitionEvent) {
        System.out.println("Sequence number of received event: " + partitionEvent.getData().getSequenceNumber());
        this.subscription.cancel(); // Cancels the subscription. No further event is received.
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("Error receiving events: " + throwable.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("Successfully completed receiving all events")
    }
});

Következtetés

A szálak költséges erőforrások, amelyeket nem szabad pazarolni a távoli szolgáltatáshívások válaszaira való várakozásra. A mikroszolgáltatási architektúrák bevezetésének növekedésével az erőforrások hatékony skálázásának és használatának szükségessége létfontosságúvá válik. Az aszinkron API-k akkor előnyösek, ha hálózathoz kötött műveletek vannak. Az Azure SDK for Java számos API-t kínál az aszinkron műveletekhez a rendszererőforrások maximalizálása érdekében. Próbálja ki az aszinkron ügyfeleket.

Az adott feladatoknak leginkább megfelelő operátorokról további információt a Reactor 3 referencia-útmutatójábantalál.

Következő lépések

Most, hogy jobban megismerte a különböző aszinkron programozási fogalmakat, fontos, hogy megtanulja, hogyan lehet iterálni az eredményeket. A legjobb iterációs stratégiákról és a lapozás működésének részleteiről további információt a Java Azure SDK-ban található Pagination és iteráció című témakörben talál.

  • Last updated on