streamalism.org

1. Bemutatkozás

Ebben a cikkben megnézzük a Netty-et - egy aszinkron eseményvezérelt hálózati alkalmazás keretrendszert.

A Netty fő célja az NIO-ra (vagy esetleg NIO.2-re) épülő, nagy teljesítményű protokollkiszolgálók kiépítése a hálózat és az üzleti logika összetevőinek elválasztásával és laza összekapcsolásával. Megvalósíthat egy széles körben ismert protokollt, például a HTTP-t, vagy a saját sajátos protokollját.

2. Alapfogalmak

A Netty nem blokkoló keretrendszer. Ez nagy teljesítményhez vezet az IO blokkolásához képest. A nem blokkoló IO megértése elengedhetetlen a Netty alapvető alkotóelemeinek és kapcsolataik megértéséhez.

2.1. Csatorna

Csatorna a Java NIO alapja. Nyílt kapcsolatot jelent, amely képes IO műveletekre, például olvasásra és írásra.

2.2. Jövő

Minden IO művelet a Csatorna Nettyben nem blokkoló.

Ez azt jelenti, hogy minden művelet a hívás után azonnal visszatér. Van egy Jövő interfész a szokásos Java könyvtárban, de Netty célokra nem kényelmes - csak a Jövő a művelet befejezéséről vagy az aktuális szál letiltásáról a művelet befejezéséig.

Ezért Nettynek megvan a sajátja ChannelFuture felület. Visszahívást továbbíthatunk ChannelFuture amelyet a művelet befejeztével hívnak meg.

2.3. Események és kezelők

Netty eseményvezérelt alkalmazási paradigmát használ, így az adatfeldolgozás folyamata egy eseménylánc, amely a kezelőkön megy keresztül. Az események és a kezelők kapcsolatba hozhatók a bejövő és a kimenő adatfolyamokkal. A bejövő események a következők lehetnek:

• Csatorna aktiválása és deaktiválása
• Műveleti események olvasása
• Kivételes események
• Felhasználói események

A kimenő események egyszerűbbek, és általában a kapcsolat megnyitásával / lezárásával és az adatok írásával / öblítésével kapcsolatosak.

A Netty alkalmazások pár hálózati és alkalmazáslogikai eseményből és azok kezelőiből állnak. A csatorna eseménykezelők alap interfészei ChannelHandler és ősei ChannelOutboundHandler és ChannelInboundHandler.

A Netty hatalmas megvalósítási hierarchiát biztosít ChannelHandler. Érdemes megjegyezni azokat az adaptereket, amelyek csak üres megvalósítások, pl. ChannelInboundHandlerAdapter és ChannelOutboundHandlerAdapter. Kiterjeszthetjük ezeket az adaptereket, amikor az összes eseménynek csak egy részhalmazát kell feldolgoznunk.

Emellett számos olyan megvalósítás létezik, amelyek specifikus protokollokat tartalmaznak, például a HTTP-t, pl. HttpRequestDecoder, HttpResponseEncoder, HttpObjectAggregator. Jó lenne Netty Javadoc-jában megismerkedni velük.

2.4. Kódolók és dekóderek

Ahogy a hálózati protokollal dolgozunk, el kell végeznünk az adatok sorosítását és a deserializációt. Erre a célra Netty bemutatja a ChannelInboundHandler mert dekóderek amelyek képesek a bejövő adatok dekódolására. A legtöbb dekóder alaposztálya az ByteToMessageDecoder.

A kimenő adatok kódolásához a Netty kiterjeszti a ChannelOutboundHandler hívott kódolók. MessageToByteEncoder az alapja a legtöbb kódoló megvalósításnak. Konvertálhatjuk az üzenetet bájtsorrendből Java objektummá és fordítva kódolókkal és dekóderekkel.

3. Példa szerver alkalmazásra

Hozzunk létre egy olyan projektet, amely egy egyszerű protokoll-kiszolgálót képvisel, amely fogad egy kérést, elvégez egy számítást és választ küld.

3.1. Függőségek

Először is meg kell adnunk a Netty-függőséget pom.xml:

 io.netty netty-all 4.1.10.Végső 

A legújabb verziót megtalálhatjuk a Maven Central oldalon.

3.2. Adatmodell

A kérelem adatosztályának a következő szerkezete lenne:

public class RequestData {private int intValue; privát String stringValue; // szabványos mérőeszközök és beállítók}

Tegyük fel, hogy a szerver megkapja a kérést és visszaadja a intValue megszorozva 2-vel. A válasz egyetlen int értékkel rendelkezik:

public class ResponseData {private int intValue; // szabványos mérőeszközök és beállítók}

3.3. Kérje a dekódert

Most kódolókat és dekódereket kell létrehoznunk a protokollüzeneteinkhez.

meg kell jegyezni, hogy A Netty a socket vételi pufferrel működik, amelyet nem sorként, hanem csak egy csomó bájtként ábrázolnak. Ez azt jelenti, hogy a bejövő kezelőnket akkor lehet hívni, amikor a szerver nem fogadja a teljes üzenetet.

A feldolgozás előtt meg kell győződnünk arról, hogy a teljes üzenetet megkaptuk és ennek sokféle módja van.

Először is létrehozhatunk ideiglenes ByteBuf és egészítse ki az összes bejövő bájtot, amíg meg nem kapjuk a szükséges bájtmenetet:

public class SimpleProcessingHandler kiterjeszti a ChannelInboundHandlerAdapter {private ByteBuf tmp; @Orride public void handlerAdded (ChannelHandlerContext ctx) {System.out.println ("Handler hozzáadva"); tmp = ctx.alloc (). puffer (4); } @Orride public void handlerRemoved (ChannelHandlerContext ctx) {System.out.println ("Handler eltávolítva"); tmp.release (); tmp = null; } @Orride public void channelRead (ChannelHandlerContext ctx, Object msg) {ByteBuf m = (ByteBuf) msg; tmp.writeBytes (m); m. kiadás (); if (tmp.readableBytes ()> = 4) {// kérelem feldolgozása RequestData requestData = új RequestData (); requestData.setIntValue (tmp.readInt ()); ResponseData responseData = új ResponseData (); responseData.setIntValue (requestData.getIntValue () * 2); ChannelFuture future = ctx.writeAndFlush (responseData); future.addListener (ChannelFutureListener.CLOSE); }}}

A fenti példa kissé furcsának tűnik, de segít megérteni Netty működését. Kezelőnk minden módszerét akkor hívjuk meg, amikor a megfelelő esemény bekövetkezik. Tehát a kezelő hozzáadásakor inicializáljuk a puffert, feltöltöttük az új bájtok fogadására vonatkozó adatokkal, és amikor elég adatot kapunk, elkezdjük feldolgozni.

Szándékosan nem használtunk a stringValue - az ilyen módon történő dekódolás feleslegesen bonyolult lenne. Ezért nyújt a Netty hasznos dekóder osztályokat, amelyek megvalósításai ChannelInboundHandler: ByteToMessageDecoder és ReplayingDecoder.

Amint azt fentebb megjegyeztük, létrehozhatunk egy csatornafeldolgozó folyamatot a Netty-vel. Tehát dekóderünket első kezelőként helyezhetjük el, és a feldolgozási logika kezelője utána jöhet.

A RequestData dekódolója a következő jelenik meg:

a public class RequestDecoder kiterjeszti a ReplayingDecoder {private final Charset charset = Charset.forName ("UTF-8"); A @Orride védett érvénytelen dekódolás (ChannelHandlerContext ctx, ByteBuf be, List out) dobja a {RequestData data = new RequestData () kivételeket; data.setIntValue (in.readInt ()); int strLen = in.readInt (); data.setStringValue (in.readCharSequence (strLen, charset) .toString ()); out.add (adatok); }}

A dekóder ötlete meglehetősen egyszerű. A. Implementációját használja ByteBuf amely kivételt vet, ha nincs elég adat a pufferben az olvasási művelethez.

Amikor a kivétel elkapta, a puffert visszatekerjük az elejére, és a dekóder új adatokra vár. A dekódolás leáll, amikor a ki a lista nem üres dekódolni végrehajtás.

3.4. Válaszkódoló

A. Dekódolása mellett RequestData kódolnunk kell az üzenetet. Ez a művelet egyszerűbb, mert a teljes üzenetadattal rendelkezünk, amikor az írási művelet megtörténik.

Írhatunk adatokat Csatorna fő kezelőnkben, vagy különválaszthatjuk a logikát, és létrehozhatunk egy kiterjesztő kezelőt MessageToByteEncoder ami elkapja az írást ResponseData művelet:

public class ResponseDataEncoder kiterjeszti a MessageToByteEncoder {@Orride védett void kódolást (ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) dobja a Kivételt {out.writeInt (msg.getIntValue ()); }}

3.5. Kérelem feldolgozása

Mivel a dekódolást és a kódolást külön kezelőkben hajtottuk végre, meg kell változtatnunk ProcessingHandler:

public class ProcessingHandler kiterjeszti a ChannelInboundHandlerAdapter {@Override public void channelRead (ChannelHandlerContext ctx, Object msg) dobja a Kivétel {RequestData requestData = (RequestData) üzenetet; ResponseData responseData = új ResponseData (); responseData.setIntValue (requestData.getIntValue () * 2); ChannelFuture future = ctx.writeAndFlush (responseData); future.addListener (ChannelFutureListener.CLOSE); System.out.println (requestData); }}

3.6. Server Bootstrap

Most rakjuk össze az egészet és futtassuk a szerverünket:

nyilvános osztály NettyServer {privát int port; // konstruktor public static void main (String [] args) dobja az {int port = args.length> 0 kivételt? Integer.parseInt (args [0]); : 8080; új NettyServer (port) .run (); } public void run () dobja a Kivételt {EventLoopGroup bossGroup = new NioEventLoopGroup (); EventLoopGroup workerGroup = új NioEventLoopGroup (); próbáld ki a {ServerBootstrap b = new ServerBootstrap (); b.group (bossGroup, workerGroup) .channel (NioServerSocketChannel.class) .childHandler (new ChannelInitializer () {@Orride public void initChannel (SocketChannel ch) dobja a Kivétel {ch.pipeline (). addLast (new RequestDecoder (), új ProcessingHandler ());}}). opció (ChannelOption.SO_BACKLOG, 128) .childOption (ChannelOption.SO_KEEPALIVE, true); ChannelFuture f = b.bind (port) .sync (); f.csatorna (). closeFuture (). szinkron (); } végül {workerGroup.shutdownGracately (); bossGroup.shutdownGracian (); }}}

A fenti szerver bootstrap példában használt osztályok részletei megtalálhatók a Javadoc-ban. A legérdekesebb rész ez a sor:

ch.pipeline (). addLast (új RequestDecoder (), új ResponseDataEncoder (), új ProcessingHandler ());

Itt meghatározzuk a bejövő és a kimenő kezelőket, amelyek a kéréseket és a kimenetet a megfelelő sorrendben dolgozzák fel.

4. Kliens alkalmazás

Az ügyfélnek fordított kódolást és dekódolást kell végrehajtania, ezért rendelkeznünk kell egy RequestDataEncoder és ResponseDataDecoder:

public class RequestDataEncoder kiterjeszti a MessageToByteEncoder {private final Charset charset = Charset.forName ("UTF-8"); A @Orride védett void kódolás (ChannelHandlerContext ctx, RequestData msg, ByteBuf out) dobja a Kivételt {out.writeInt (msg.getIntValue ()); out.writeInt (msg.getStringValue (). hossz ()); out.writeCharSequence (msg.getStringValue (), karakterkészlet); }}

public class ResponseDataDecoder kiterjeszti a ReplayingDecoder {@Orride védett void dekódolást (ChannelHandlerContext ctx, ByteBuf be, List out) dobja a Kivételt {ResponseData data = new ResponseData (); data.setIntValue (in.readInt ()); out.add (adatok); }}

Meg kell határoznunk a ClientHandler amely elküldi a kérést és megkapja a választ a szervertől:

public class ClientHandler kiterjeszti a ChannelInboundHandlerAdapter {@Override public void channelActive (ChannelHandlerContext ctx) dobja a Kivételt {RequestData msg = new RequestData (); msg.setIntValue (123); msg.setStringValue ("minden munka és játék nélkül a fiú unalmas fiúvá válik"); ChannelFuture future = ctx.writeAndFlush (msg); } @Orride public void channelRead (ChannelHandlerContext ctx, Object msg) a {System.out.println ((ResponseData) msg) kivételt dobja; ctx.close (); }}

Most indítsuk el a klienst:

public class NettyClient {public static void main (String [] args) dobja a Kivételt {String host = "localhost"; int port = 8080; EventLoopGroup workerGroup = új NioEventLoopGroup (); próbáld ki a {Bootstrap b = new Bootstrap () -t; b.csoport (workerGroup); b.csatorna (NioSocketChannel.class); b.option (ChannelOption.SO_KEEPALIVE, true); b.handler (new ChannelInitializer () {@Orride public void initChannel (SocketChannel ch) dobja a Kivételt {ch.pipeline (). addLast (új RequestDataEncoder (), új ResponseDataDecoder (), új ClientHandler ());}}); ChannelFuture f = b.csatlakozás (gazdagép, port) .sync (); f.csatorna (). closeFuture (). szinkron (); } végül {workerGroup.shutdownGracately (); }}}

Mint láthatjuk, a kiszolgáló rendszerindításával sok közös rész van.

Most futtathatjuk az ügyfél fő módszerét, és megnézhetjük a konzol kimenetét. Ahogy az várható volt, megkaptuk ResponseData val vel intValue egyenlő 246-tal.

5. Következtetés

Ebben a cikkben gyorsan bemutattuk Nettyt. Megmutattuk alapvető alkotóelemeit, mint pl Csatorna és ChannelHandler. Készítettünk egy egyszerű, nem blokkoló protokollkiszolgálót és egy klienst is hozzá.

Mint mindig, minden kódminta elérhető a GitHubon.