streamalism.org

1. Áttekintés

Egyszerűen fogalmazva, az MBassador az nagy teljesítményű eseménybusz, amely a közzététel-feliratkozás szemantikát használja.

Az üzeneteket egy vagy több társ felé közvetítik, anélkül, hogy előzetesen tudnánk, hány előfizető van, vagy hogyan használják az üzenetet.

2. Maven-függőség

A könyvtár használata előtt hozzá kell adnunk a nagykövet függőségét:

 net.engio nagykövet 1.3.1 

3. Alapvető eseménykezelés

3.1. Egyszerű példa

Kezdjük az üzenet közzétételének egyszerű példájával:

magán MBassador diszpécser = új MBassador (); privát String messageString; @ Mielőtt nyilvános void PreparTests () {diszpécser.subscribe (this); } @Test public void whenStringDispatched_thenHandleString () {dispatcher.post ("TestString"). Now (); assertNotNull (messageString); assertEquals ("TestString", messageString); } @Handler public void handleString (String üzenet) {messageString = üzenet; } 

Ennek a tesztosztálynak a tetején láthatjuk a MBassador alapértelmezett konstruktorával. Ezután a @Előtt módszer, hívjuk Iratkozz fel() és átad egy hivatkozást magának az osztálynak.

Ban ben Iratkozz fel(), a diszpécser megvizsgálja az előfizetőt @Handler annotációk.

És az első tesztben hívunk diszpécser.post (…). most () az üzenet elküldéséhez - ami azt eredményezi handleString () hívják.

Ez a kezdeti teszt több fontos fogalmat mutat be. Bármi Tárgy lehet előfizető, amennyiben egy vagy több metódussal rendelkezik @Handler. Az előfizetőnek tetszőleges számú kezelője lehet.

Olyan tesztobjektumokat használunk, amelyek az egyszerűség kedvéért feliratkoznak magukra, de a legtöbb gyártási szcenárióban az üzenetek diszpécserei a fogyasztóktól eltérő osztályokba tartoznak.

A Handler módszereknek csak egy bemeneti paraméterük van - az üzenet, és nem dobhatnak be semmilyen bejelölt kivételt.

Hasonló a Iratkozz fel() metódus, a post módszer bármelyiket elfogadja Tárgy. Ez Tárgy az előfizetőknek kézbesítik.

Amikor egy üzenetet közzéteszünk, azt minden hallgatónak eljuttatjuk, aki feliratkozott az üzenet típusára.

Vegyünk fel egy másik üzenetkezelőt, és küldjünk egy másik típusú üzenetet:

privát egész számüzenetInteger; @Test public void whenIntegerDispatched_thenHandleInteger () {diszpécser.post (42). Most (); assertNull (messageString); assertNotNull (messageInteger); assertTrue (42 == messageInteger); } @Handler public void handleInteger (Egész szám) {messageInteger = üzenet; } 

A várakozásoknak megfelelően, amikor elküldjükan Egész szám, handleInteger () hívják, és handleString () nem. Egyetlen diszpécser használható több üzenettípus küldésére.

3.2. Halott üzenetek

Tehát hová kerül az üzenet, ha nincs rá kezelő? Adjunk hozzá egy új eseménykezelőt, majd küldjünk egy harmadik üzenet típust:

private Object deadEvent; @Test public void whenLongDispatched_thenDeadEvent () {diszpécser.post (42L) .now (); assertNull (messageString); assertNull (messageInteger); assertNotNull (deadEvent); assertTrue (deadEvent instanceof Long); assertTrue (42L == (Long) deadEvent); } @Handler public void handleDeadEvent (DeadMessage üzenet) {deadEvent = message.getMessage (); } 

Ebben a tesztben a Hosszú an helyett Egész szám. Se handleInteger () sem handleString () hívják, de handleDeadEvent () van.

Ha nincs üzenetkezelő, akkor az a-ba csomagolódik DeadMessage tárgy. Mivel hozzáadtunk egy kezelőt Halott üzenet, elfogjuk.

DeadMessage biztonságosan figyelmen kívül hagyható; ha egy alkalmazásnak nem kell nyomon követnie a halott üzeneteket, akkor hagyhatják, hogy sehova se menjenek.

4. Esemény-hierarchia használata

Küldés Húr és Egész szám események korlátozzák. Hozzunk létre néhány üzenetosztályt:

public class Message {} nyilvános osztály AckMessage kiterjeszti az üzenetet {} public class RejectMessage kiterjeszti az üzenetet {int code; // beállítók és szerelők}

Van egy egyszerű alaposztályunk és két osztály, amelyek kibővítik.

4.1. Alaposztály küldése Üzenet

Kezdjük Üzenet események:

magán MBassador diszpécser = új MBassador (); privát üzenet üzenet; privát AckMessage ackMessage; privát RejectMessage rejectMessage; @ Mielőtt nyilvános void PreparTests () {diszpécser.subscribe (this); } @Test public void whenMessageDispatched_thenMessageHandled () {dispatcher.post (new Message ()). Now (); assertNotNull (üzenet); assertNull (ackMessage); assertNull (rejectMessage); } @Handler public void handleMessage (Üzenetüzenet) {this.message = üzenet; } @Handler public void handleRejectMessage (RejectMessage message) {rejectMessage = message; } @Handler public void handleAckMessage (AckMessage üzenet) {ackMessage = üzenet; }

Fedezze fel az MBassadort - egy nagy teljesítményű pub-sub rendezvénybusz. Ez korlátoz minket a használatra üzenetek de hozzáteszi a típusbiztonság további rétegét.

Amikor elküldünk egy Üzenet, handleMessage () megkapja. A másik két kezelő nem.

4.2. Alosztályú üzenet küldése

Küldjünk egy RejectMessage:

@Test public void whenRejectDispatched_thenMessageAndRejectHandled () {dispatcher.post (new RejectMessage ()). Now (); assertNotNull (üzenet); assertNotNull (rejectMessage); assertNull (ackMessage); }

Amikor elküldünk egy RejectMessage mindkét handleRejectMessage () és handleMessage () megkapja.

Mivel RejectMessage kiterjed Üzenet, a Üzenet kezelő megkapta, a RejectMessage kezelő.

Ellenőrizzük ezt a viselkedést egy AckMessage:

@Test public void whenAckDispatched_thenMessageAndAckHandled () {dispatcher.post (new AckMessage ()). Now (); assertNotNull (üzenet); assertNotNull (ackMessage); assertNull (rejectMessage); }

Ahogy számítottunk rá, amikor küldünk egy AckMessage, mindkét handleAckMessage () és handleMessage () megkapja.

5. Üzenetek szűrése

Az üzenetek típus szerinti rendezése már most is hatékony funkció, de még jobban szűrhetjük őket.

5.1. Szűrés osztályon és alosztályon

Amikor kifüggesztettük a RejectMessage vagy AckMessage, az eseményt az adott típus eseménykezelőjében és az alaposztályban egyaránt megkaptuk.

Megoldhatjuk ezt a típusú hierarchiát azáltal, hogy elkészítjük Üzenet absztrakt és olyan osztály létrehozása, mint GenericMessage. De mi van, ha nincs ez a luxus?

Használhatunk üzenetszűrőket:

privát üzenetbázisMessage; privát üzenet subMessage; @Test public void whenMessageDispatched_thenMessageFiltered () {dispatcher.post (new Message ()). Now (); assertNotNull (baseMessage); assertNull (subMessage); } @Test public void whenRejectDispatched_thenRejectFiltered () {dispatcher.post (new RejectMessage ()). Now (); assertNotNull (subMessage); assertNull (baseMessage); } @Handler (szűrők = {@Filter (Filters.RejectSubtypes.class)}) public void handleBaseMessage (Üzenetüzenet) {this.baseMessage = üzenet; } @Handler (szűrők = {@Filter (Filters.SubtypesOnly.class)}) public void handleSubMessage (Üzenetüzenet) {this.subMessage = üzenet; }

A szűrők paraméter a @ Kezelő az annotáció elfogadja a Osztály hogy megvalósítja IMessageFilter. A könyvtár két példát kínál:

A Szűrők.RejectSubtypes úgy tesz, ahogy a neve is mutatja: kiszűri az esetleges altípusokat. Ebben az esetben ezt látjuk RejectMessage nem kezeli handleBaseMessage ().

A Szűrők. Altípusok Csak azt is teszi, ahogy a neve is sugallja: kiszűr minden alaptípust. Ebben az esetben ezt látjuk Üzenet nem kezeli handleSubMessage ().

5.2. IMessageFilter

A Szűrők.RejectSubtypes és a Szűrők. Altípusok Csak mindkettő megvalósítja IMessageFilter.

RejectSubTypes összehasonlítja az üzenet osztályát a definiált üzenettípusokkal, és csak olyan üzeneteken keresztül engedélyezi az üzenetet, amelyek megegyeznek az egyik típusával, szemben az alosztályokkal.

5.3. Szűrés feltételekkel

Szerencsére van egy egyszerűbb módszer az üzenetek szűrésére. Az MBassador a Java EL kifejezések egy részhalmazát támogatja az üzenetek szűrésének feltételeiként.

Szűrjük a Húr üzenet a hossza alapján:

privát String testString; @Test public void whenLongStringDispatched_thenStringFiltered () {dispatcher.post ("foobar!"). Now (); assertNull (testString); } @Handler (condition = "msg.length () <7") public void handleStringMessage (String üzenet) {this.testString = üzenet; }

A „foobar!” az üzenet hét karakter hosszú és szűrve van. Küldjünk rövidebbet Húr:

 @Test public void whenShortStringDispatched_thenStringHandled () {dispatcher.post ("foobar"). Now (); assertNotNull (testString); }

Most a „foobar” csak hat karakter hosszú és átkerül.

A mi RejectMessage mezőt tartalmaz egy hozzáféréssel. Írjunk ehhez szűrőt:

privát RejectMessage rejectMessage; @Test public void whenWrongRejectDispatched_thenRejectFiltered () {RejectMessage testReject = new RejectMessage (); testReject.setCode (-1); diszpécser.post (testReject). most (); assertNull (rejectMessage); assertNotNull (subMessage); assertEquals (-1, ((RejectMessage) alMessage) .getCode ()); } @Handler (feltétel = "msg.getCode ()! = -1") public void handleRejectMessage (RejectMessage rejectMessage) {this.rejectMessage = rejectMessage; }

Itt ismét lekérdezhetünk egy metódust egy objektumon, és vagy szűrhetjük az üzenetet, vagy sem.

5.4. Szűrt üzenetek rögzítése

Hasonló DeadEvents, érdemes szűrt üzeneteket rögzíteni és feldolgozni. Van egy külön mechanizmus a szűrt események rögzítésére is. A szűrt eseményeket a „holt” eseményektől eltérően kezelik.

Írjunk egy tesztet, amely ezt szemlélteti:

privát String testString; privát FilteredMessage filteredMessage; privát DeadMessage deadMessage; @Test public void whenLongStringDispatched_thenStringFiltered () {dispatcher.post ("foobar!"). Now (); assertNull (testString); assertNotNull (filteredMessage); assertTrue (filteredMessage.getMessage () String példánya); assertNull (deadMessage); } @Handler (condition = "msg.length () <7") public void handleStringMessage (String üzenet) {this.testString = üzenet; } @Handler public void handleFilterMessage (FilteredMessage message) {this.filteredMessage = message; } @Handler public void handleDeadMessage (DeadMessage deadMessage) {this.deadMessage = deadMessage; } 

A hozzáadásával FilteredMessage kezelő, nyomon tudjuk követni Húrok amelyeket hosszuk miatt szűrnek. A filterMessage túl hosszút tartalmaz Húr míg deadMessage maradványok nulla.

6. Aszinkron üzenetküldés és -kezelés

Eddig valamennyi példánk szinkron üzenetküldést alkalmazott; amikor felhívtunk post.now () az üzeneteket minden kezelőhöz ugyanazon a szálon juttattuk el, amelyet hívtunk post () tól től.

6.1. Aszinkron diszpécser

A MBassador.post () visszaad egy SyncAsyncPostCommand parancsot. Ez az osztály számos módszert kínál, többek között:

• Most() - szinkron módon küldje el az üzeneteket; a hívás az összes üzenet kézbesítéséig blokkolódik
• aszinkron módon () - aszinkron módon hajtja végre az üzenet közzétételét

Használjunk aszinkron diszpécsert egy minta osztályban. Az Awaitility használatával ezekben a tesztekben egyszerűsítjük a kódot:

magán MBassador diszpécser = új MBassador (); privát String testString; private AtomicBoolean ready = új AtomicBoolean (hamis); @Test public void whenAsyncDispatched_thenMessageReceived () {diszpécser.post ("foobar"). Aszinkron (); várjon (). amígAtomic (kész, egyenlőTo (igaz)); assertNotNull (testString); } @Handler public void handleStringMessage (String üzenet) {this.testString = üzenet; kész.készlet (igaz); }

Hívjuk aszinkron módon () ebben a tesztben, és használjon egy AtomicBoolean mint zászlóval várják() hogy megvárjam, amíg a kézbesítési szál eljuttatja az üzenetet.

Ha megjegyezzük a hívást várják(), kockáztatjuk a teszt kudarcát, mert ellenőrizzük testString mielőtt a szállítási szál befejeződik.

6.2. Aszinkron kezelő meghívása

Az aszinkron diszpécselés lehetővé teszi az üzenetszolgáltató számára, hogy visszatérjen az üzenetek feldolgozásához, mielőtt az üzeneteket minden kezelőhöz eljuttatnák, de továbbra is rendre hívja az összes kezelőt, és mindegyik kezelőnek meg kell várnia az előző befejezését.

Ez problémákat okozhat, ha az egyik kezelő drága műveletet hajt végre.

Az MBassador egy mechanizmust biztosít az aszinkron kezelő meghívásához. Az erre konfigurált kezelők üzeneteket kapnak a szálukban:

magán Integer tesztInteger; privát String invocationThreadName; private AtomicBoolean ready = új AtomicBoolean (hamis); @Test public void whenHandlerAsync_thenHandled () {diszpécser.post (42) .now (); várjon (). amígAtomic (kész, egyenlőTo (igaz)); assertNotNull (testInteger); assertFalse (Thread.currentThread (). getName (). egyenlő (invocationThreadName)); } @Handler (szállítás = Invoke.Aszinkron módon) public void handleIntegerMessage (Integer message) {this.invocationThreadName = Thread.currentThread (). GetName (); this.testInteger = üzenet; kész.készlet (igaz); }

A kezelők aszinkron meghívást kérhetnek a kézbesítés = Invoke.Aszinkron módon ingatlan a Handler annotáció. Tesztünk során ezt ellenőrizzük a cérna nevek a diszpécser módszerben és a kezelő.

7. Az MBassador testreszabása

Eddig az MBassador egy példányát használtuk az alapértelmezett konfigurációval. A diszpécser viselkedését annotációkkal lehet módosítani, hasonlóan az eddigiekhez; bemutatunk még néhányat az oktatóanyag befejezéséhez.

7.1. Kivételek kezelése

A kezelők nem definiálhatják az ellenőrzött kivételeket. Ehelyett a diszpécser számára biztosítható egy IPublicationErrorHandler érvként a konstruktőre:

az MBassadorConfigurationTest nyilvános osztály megvalósítja az IPublicationErrorHandler {magán MBassador diszpécsert; privát String messageString; privát Throwable errorCause; @ Mielőtt nyilvános void PreparTests () {diszpécser = new MBassador (this); diszpécser.feliratkozás (erre); } @Test public void whenErrorOccurs_thenErrorHandler () {dispatcher.post ("Error"). Now (); assertNull (messageString); assertNotNull (errorCause); } @Test public void whenNoErrorOccurs_thenStringHandler () {dispatcher.post ("Error"). Now (); assertNull (errorCause); assertNotNull (messageString); } @Handler public void handleString (String üzenet) {if ("Hiba" .egyenlő (üzenet)) {dobjon új hibát ("BOOM"); } messageString = üzenet; } @Orride public void handleError (PublicationError error) {errorCause = error.getCause (). GetCause (); }}

Mikor handleString () dob egy Hiba, menti errorCause.

7.2. Kezelő prioritása

Kezelők fordított sorrendben hívják meg, ahogyan hozzáadják őket, de nem erre a viselkedésre szeretnénk támaszkodni. Még akkor is, ha képesek vagyunk kezelőket hívni a szálaikban, akkor is tudnunk kell, hogy milyen sorrendben hívják őket.

Kifejezetten beállíthatjuk a kezelő prioritását:

privát LinkedList list = új LinkedList (); @Test public void whenRejectDispatched_thenPriorityHandled () {dispatcher.post (new RejectMessage ()). Now (); // Az elemeknek ki kell ugraniuk () az assertTrue (1 == list.pop ()) fordított prioritású sorrendben; assertTrue (3 == list.pop ()); assertTrue (5 == list.pop ()); } @Handler (prioritás = 5) public void handleRejectMessage5 (RejectMessage rejectMessage) {list.push (5); } @Handler (prioritás = 3) public void handleRejectMessage3 (RejectMessage rejectMessage) {list.push (3); } @Handler (prioritás = 2, rejectSubtypes = true) public void handleMessage (Message rejectMessage) logger.error ("3. kezelő elutasítása"); list.push (3); } @Handler (prioritás = 0) public void handleRejectMessage0 (RejectMessage rejectMessage) {list.push (1); } 

A kezelőket a legmagasabb prioritásról a legalacsonyabbra hívják. Az alapértelmezett prioritású (nulla) kezelőket utoljára hívják. Látjuk, hogy a kezelő számokat ad pop() fordított sorrendben.

7.3. Elutasítja az altípusokat, az egyszerű utat

Mi történt handleMessage () a fenti tesztben? Nem kell használnunk RejectSubTypes.class az altípusaink szűrésére.

RejectSubTypes egy logikai jelző, amely ugyanazt a szűrést biztosítja, mint az osztály, de jobb teljesítményt nyújt, mint a IMessageFilter végrehajtás.

A szűrőalapú megvalósítást azonban továbbra is csak az altípusok elfogadásához kell használnunk.

8. Következtetés

Az MBassador egy egyszerű és egyszerű könyvtár az üzenetek objektumok közötti továbbításához. Az üzenetek különféle módon rendezhetők, és szinkronosan vagy aszinkron módon küldhetők el.

És mint mindig, a példa ebben a GitHub projektben is elérhető.