streamalism.org

1. Bemutatkozás

Ebben az oktatóanyagban megismerhetjük a tavaszi integráció Java DSL-jét az alkalmazásintegrációk létrehozásához.

Fogjuk a Bevezetés a tavaszi integrációba beépített fájlmozgató integrációt, és helyette a DSL-t használjuk.

2. Függőségek

A Spring Integration Java DSL a Spring Integration Core része.

Tehát hozzáadhatjuk ezt a függőséget:

 org.springframework.integration spring-integration-core 5.0.6.KÖZLEMÉNY 

A fájlmozgató alkalmazásunkhoz pedig szükségünk lesz a tavaszi integrációs fájlra is:

 org.springframework.integration spring-integration-file 5.0.6.KÖZLEMÉNY 

3. Tavaszi integráció Java DSL

A Java DSL előtt a felhasználók XML-ben konfigurálják a Spring Integration összetevőket.

A DSL bemutat néhány folyékony építőt, amelyekből egyszerűen létrehozhatunk egy teljes tavaszi integrációs folyamatot pusztán Java-ban.

Tehát tegyük fel, hogy létre akartunk hozni egy csatornát, amely a csövön keresztül érkező összes adatot nagybetűsíti.

A múltban talán megtettük:

És most megtehetjük:

@Bean public IntegrationFlow upcaseFlow () {return IntegrationFlows.from ("input") .transform (String :: toUpperCase) .get (); }

4. A fájlmozgató alkalmazás

A fájlmozgató integráció megkezdéséhez szükségünk lesz néhány egyszerű építőelemre.

4.1. Integrációs folyamat

Az első építőelem, amire szükségünk van, egy integrációs folyamat, amelyet a IntegrationFlows építész:

IntegrationFlows.from (...)

tól től többféle lehet, de ebben az oktatóanyagban csak hármat nézünk meg:

• MessageSources
• MessageChannels és
• Húrs

Rövidesen beszélünk mindháromról.

Miután felhívtunk tól től, néhány testreszabási módszer már elérhető számunkra:

IntegrationFlow flow = IntegrationFlows.from (sourceDirectory ()) .filter (onlyJpgs ()) .handle (targetDirectory ()) // további összetevők hozzáadása .get ();

Végül, IntegrationFlows mindig létrehoz egy példányát IntegrationFlow, amely bármely tavaszi integrációs alkalmazás végterméke.

Ez a beviteli minta, a megfelelő átalakítások végrehajtása és az eredmények kibocsátása alapvető az összes tavaszi integrációs alkalmazás számára.

4.2. Bemeneti forrás leírása

Először is, a fájlok áthelyezéséhez meg kell jelölnünk az integrációs folyamatunknak, hogy hol kell keresnie őket, és ehhez szükségünk van egy MessageSource:

@Bean public MessageSource sourceDirectory () {// .. üzenetforrás létrehozása}

Egyszerűen fogalmazva: a MessageSource olyan hely, ahonnan az alkalmazáson kívüli üzenetek érkezhetnek.

Pontosabban, szükségünk van valamire, ami képes alkalmazkodni azt a külső forrást a tavaszi üzenetküldési ábrázolásba. És mivel ez alkalmazkodás arra összpontosít bemenet, ezeket gyakran hívják Bemeneti csatornaadapterek.

A tavaszi integrációs fájl a függőség olyan bemeneti csatornaadaptert ad nekünk, amely nagyszerű a felhasználási esetünkhöz: FileReadingMessageSource:

@Bean public MessageSource sourceDirectory () {FileReadingMessageSource messageSource = új FileReadingMessageSource (); messageSource.setDirectory (új fájl (INPUT_DIR)); return messageSource; }

Itt, a mi FileReadingMessageSource által megadott könyvtárat fogja olvasni INPUT_DIR és létrehoz egy MessageSource ebből.

Adjuk meg ezt forrásként a IntegrationFlows.from meghívás:

IntegrationFlows.from (sourceDirectory ());

4.3. Bemeneti forrás konfigurálása

Ha hosszú életű alkalmazásra gondolunk, valószínűleg azt akarjuk, hogy észrevegyük a fájlokat, amikor bejönnek, nem csak az indításkor már meglévő fájlokat helyezheti át.

Ennek megkönnyítése érdekében tól től extra is vehet konfigurátorok a bemeneti forrás további testreszabása:

IntegrationFlows.from (sourceDirectory (), konfiguráló -> configurer.poller (Pollers.fixedDelay (10000)));

Ebben az esetben rugalmasabbá tehetjük a bemeneti forrást, ha azt mondjuk a Spring Integration-nek, hogy 10 másodpercenként kérdezze meg ezt a forrást - ebben az esetben a fájlrendszerünket.

És ez természetesen nem csak a fájl bemeneti forrásunkra vonatkozik, ezt a közvélemény-kutatót bármelyikhez hozzáadhatjuk MessageSource.

4.4. Üzenetek szűrése bemeneti forrásból

Tegyük fel, hogy azt akarjuk, hogy a fájlokat mozgató alkalmazásunk csak bizonyos fájlokat helyezzen át, mondjuk a képfájlokat jpg kiterjesztés.

Ehhez használhatjuk GenericSelector:

@Bean public GenericSelector onlyJpgs () {return new GenericSelector () {@Orride public boolean accept (File source) {return source.getName (). EndWith (". Jpg"); }}; }

Tehát frissítsük újra az integrációs folyamatunkat:

IntegrationFlows.from (sourceDirectory ()) .filter (csakJpgs ());

Vagy, mivel ez a szűrő olyan egyszerű, helyette definiálhattuk volna lambda használatával:

IntegrationFlows.from (sourceDirectory ()) .filter (forrás -> ((Fájl) forrás) .getName (). EndWith (". Jpg"));

4.5. Üzenetek kezelése szolgáltatásaktivátorokkal

Most, hogy van egy szűrt fájllista, új helyre kell írnunk őket.

Szolgáltatás aktivátors azokhoz fordulunk, amikor a tavaszi integráció eredményeire gondolunk.

Használjuk a FileWritingMessageHandler szolgáltatás aktivátor tavaszi integrációs fájl:

@Bean public MessageHandler targetDirectory () {FileWritingMessageHandler handler = új FileWritingMessageHandler (új File (OUTPUT_DIR)); handler.setFileExistsMode (FileExistsMode.REPLACE); handler.setExpectReply (hamis); visszatérő kezelő; }

Itt, a mi FileWritingMessageHandler mindegyiket megírja Üzenet hasznos terhelést kap OUTPUT_DIR.

Ismét frissítsük:

IntegrationFlows.from (sourceDirectory ()) .filter (onlyJpgs ()) .handle (targetDirectory ());

És figyeld meg egyébként a setExpectReply. Mert az integrációs folyamatok lehetnekkétirányú, ez az invokáció azt jelzi, hogy ez a bizonyos cső egyirányú.

4.6. Integrációs folyamatunk aktiválása

Amikor hozzáadtuk az összes szükséges elemet regisztrálja a mi IntegrationFlow babként aktiválásához:

@Bean public IntegrationFlow fileMover () {return IntegrationFlows.from (sourceDirectory (), c -> c.poller (Pollers.fixedDelay (10000))) .filter (onlyJpgs ()) .handle (targetDirectory ()) .get () ; }

A kap módszer kivonat egy IntegrationFlow például, hogy regisztrálnunk kell tavaszi babként.

Amint az alkalmazás-kontextusunk betöltődik, az összes elemünk benne van IntegrationFlow aktiválódik.

És most az alkalmazásunk megkezdi a fájlok áthelyezését a forráskönyvtárból a célkönyvtárba.

5. További alkatrészek

DSL alapú fájlmozgató alkalmazásunkban létrehoztunk egy bejövő csatornaadaptert, egy üzenetszűrőt és egy szolgáltatásaktivátort.

Nézzünk meg néhány más általános tavaszi integrációs összetevőt, és nézzük meg, hogyan használhatnánk őket.

5.1. Üzenetcsatornák

Mint korábban említettük, a Üzenetcsatorna az áramlás inicializálásának egy másik módja:

IntegrationFlows.from ("anyChannel")

Ezt úgy olvashatjuk, hogy „kérjük, keresse meg vagy hozzon létre egy csatornababot, amelynek neve anyChannel. Ezután olvassa el az összes betáplált adatot anyChannel más áramlásoktól. ”

De valójában ennél általánosabb célú.

Egyszerűen fogalmazva: egy csatorna elvonja a gyártókat a fogyasztóktól, és felfoghatjuk Java-ként Sor. Csatorna beilleszthető az áramlás bármely pontjára.

Tegyük fel például, hogy fontossági sorrendben szeretnénk megadni a fájlokat, amikor az egyik könyvtárból a másikba kerülnek:

@Bean public PriorityChannel betűrendben () {return new PriorityChannel (1000, (balra, jobbra) -> ((Fájl) left.getPayload ()). GetName (). CompareTo ((((File) right.getPayload ()). GetName ())); }

Ezután beilleszthetünk egy meghívást ide csatorna áramlásunk között:

@Bean public IntegrationFlow fileMover () {return IntegrationFlows.from (sourceDirectory ()) .filter (onlyJpgs ()) .channel ("ábécé szerint") .handle (targetDirectory ()) .get (); }

Több tucat csatorna közül lehet választani, néhány praktikusabb egyidejűség, auditálás vagy közbenső kitartás (gondoljuk Kafka vagy JMS pufferekre).

Ezenkívül a csatornák erősek lehetnek, ha velük kombinálják Híds.

5.2. Híd

Amikor akarjuk két csatornát kombinál, használjuk a Híd.

Képzeljük el, hogy ahelyett, hogy közvetlenül egy kimeneti könyvtárba írtunk volna, a fájlokat mozgató alkalmazásunkat egy másik csatornára írtuk:

@Bean public IntegrationFlow fileReader () {return IntegrationFlows.from (sourceDirectory ()) .filter (onlyJpgs ()) .channel ("holdingTank") .get (); }

Most, mert egyszerűen egy csatornára írtuk, áthidalhatunk onnan más áramlásokra.

Hozzunk létre egy hidat, amely lekérdezi a tárolótartályunkat az üzenetek után, és azokat egy rendeltetési helyre írja:

@Bean public IntegrationFlow fileWriter () {return IntegrationFlows.from ("holdingTank") .bridge (e -> e.poller (Pollers.fixedRate (1, TimeUnit.SECONDS, 20))) .handle (targetDirectory ()) .get (); }

Ismét, mivel egy köztes csatornára írtunk, most hozzáadhatunk egy másik folyamatot amely ugyanazokat a fájlokat veszi és más ütemben írja:

@Bean public IntegrationFlow anotherFileWriter () {return IntegrationFlows.from ("holdingTank") .bridge (e -> e.poller (Pollers.fixedRate (2, TimeUnit.SECONDS, 10))) .handle (anotherTargetDirectory ()) .get (); }

Mint láthatjuk, az egyes hidak vezérelhetik a lekérdezési konfigurációt a különböző kezelők számára.

Amint betöltődik az alkalmazáskörnyezetünk, most egy bonyolultabb alkalmazással rendelkezünk, amely megkezdi a fájlok áthelyezését a forráskönyvtárból két célkönyvtárba.

6. Következtetés

Ebben a cikkben különféle módszereket láttunk a Spring Integration Java DSL használatára különböző integrációs csővezetékek kiépítésére.

Lényegében egy korábbi oktatóanyagból sikerült újra létrehoznunk a fájlmozgató alkalmazást, ezúttal tiszta java használatával.

Megnéztünk néhány további elemet, például csatornákat és hidakat.

Az oktatóanyagban használt teljes forráskód elérhető a Github oldalon.