streamalism.org

1. Introduktion

I denne artikel skal vi se på Netty - en asynkron begivenhedsdrevet netværksapplikationsramme.

Hovedformålet med Netty er at opbygge højtydende protokolservere baseret på NIO (eller muligvis NIO.2) med adskillelse og løs kobling af netværket og forretningslogikkomponenter. Det implementerer muligvis en bredt kendt protokol, såsom HTTP, eller din egen specifikke protokol.

2. Kernebegreber

Netty er en ikke-blokerende ramme. Dette fører til høj kapacitet sammenlignet med blokering af IO. Forståelse af ikke-blokerende IO er afgørende for at forstå Netts kernekomponenter og deres forhold.

2.1. Kanal

Kanal er basen for Java NIO. Det repræsenterer en åben forbindelse, der er i stand til IO-operationer såsom læsning og skrivning.

2.2. Fremtid

Hver IO-operation på en Kanal i Netty er ikke-blokering.

Dette betyder, at hver operation returneres straks efter opkaldet. Der er en Fremtid interface i standard Java-biblioteket, men det er ikke praktisk til Netty-formål - vi kan kun spørge Fremtid om afslutningen af ​​operationen eller for at blokere den aktuelle tråd, indtil operationen er færdig.

Derfor Netty har sin egen ChannelFuture interface. Vi kan give et tilbagekald til ChannelFuture som kaldes efter operationens afslutning.

2.3. Begivenheder og håndterere

Netty bruger et begivenhedsdrevet applikationsparadigme, så rørledningen til databehandlingen er en kæde af begivenheder, der gennemgår håndterere. Begivenheder og håndterere kan relateres til den indgående og udgående datastrøm. Indgående begivenheder kan være følgende:

• Kanalaktivering og deaktivering
• Læs driftshændelser
• Undtagelsesbegivenheder
• Brugerbegivenheder

Udgående hændelser er enklere og er generelt relateret til åbning / lukning af en forbindelse og skrivning / skylning af data.

Net-applikationer består af et par netværks- og applikationslogiske begivenheder og deres håndterere. Basisgrænsefladerne for kanalhændelseshåndtererne er ChannelHandler og dets forfædre ChannelOutboundHandler og ChannelInboundHandler.

Netty giver et stort hierarki af implementeringer af ChannelHandler. Det er værd at bemærke adapterne, der bare er tomme implementeringer, f.eks. ChannelInboundHandlerAdapter og ChannelOutboundHandlerAdapter. Vi kunne udvide disse adaptere, når vi kun har brug for at behandle en delmængde af alle begivenheder.

Der er også mange implementeringer af specifikke protokoller såsom HTTP, f.eks. HttpRequestDecoder, HttpResponseEncoder, HttpObjectAggregator. Det ville være godt at stifte bekendtskab med dem i Netty's Javadoc.

2.4. Kodere og dekodere

Når vi arbejder med netværksprotokollen, skal vi udføre dataserialisering og deserialisering. Til dette formål introducerer Netty specielle udvidelser af ChannelInboundHandler til dekodere som er i stand til at afkode indgående data. Basisklassen for de fleste dekodere er ByteToMessageDecoder.

Til kodning af udgående data har Netty udvidelser af ChannelOutboundHandler hedder kodere. MessageToByteEncoder er basen for de fleste encoderimplementeringer. Vi kan konvertere meddelelsen fra bytesekvens til Java-objekt og omvendt med kodere og dekodere.

3. Eksempel på serverapplikation

Lad os oprette et projekt, der repræsenterer en simpel protokolserver, der modtager en anmodning, udfører en beregning og sender et svar.

3.1. Afhængigheder

Først og fremmest er vi nødt til at give Netty afhængighed i vores pom.xml:

 io.netty netty-all 4.1.10.Final 

Vi kan finde den nyeste version på Maven Central.

3.2. Datamodel

Dataklassen for anmodning ville have følgende struktur:

offentlig klasse RequestData {private int intValue; private String stringValue; // standard getters og setter}

Lad os antage, at serveren modtager anmodningen og returnerer intValue ganget med 2. Svaret ville have den enkelte int-værdi:

offentlig klasse ResponseData {private int intValue; // standard getters og setter}

3.3. Anmod dekoder

Nu skal vi oprette kodere og dekodere til vores protokolbeskeder.

Det skal bemærkes, at Netty arbejder med socket modtage buffer, som ikke er repræsenteret som en kø, men bare som en flok bytes. Dette betyder, at vores indgående handler kan kaldes, når den fulde besked ikke modtages af en server.

Vi skal sikre os, at vi har modtaget den fulde besked inden behandlingen og der er mange måder at gøre det på.

Først og fremmest kan vi oprette en midlertidig ByteBuf og tilføj det alle indgående byte, indtil vi får den krævede mængde byte:

offentlig klasse SimpleProcessingHandler udvider ChannelInboundHandlerAdapter {privat ByteBuf tmp; @Override public void handlerAdded (ChannelHandlerContext ctx) {System.out.println ("Handler tilføjet"); tmp = ctx.alloc (). buffer (4); } @ Override public void handlerRemoved (ChannelHandlerContext ctx) {System.out.println ("Handler fjernet"); tmp.release (); tmp = null; } @ Override public void channelRead (ChannelHandlerContext ctx, Object msg) {ByteBuf m = (ByteBuf) msg; tmp.writeBytes (m); m.frigivelse (); hvis (tmp.readableBytes ()> = 4) {// anmodning behandler RequestData requestData = ny RequestData (); requestData.setIntValue (tmp.readInt ()); ResponseData responseData = ny ResponseData (); responseData.setIntValue (requestData.getIntValue () * 2); ChannelFuture fremtid = ctx.writeAndFlush (responsData); future.addListener (ChannelFutureListener.CLOSE); }}}

Eksemplet vist ovenfor ser lidt underligt ud, men hjælper os med at forstå, hvordan Netty fungerer. Hver metode til vores handler kaldes, når den tilsvarende begivenhed finder sted. Så vi initialiserer bufferen, når handleren tilføjes, udfylder den med data om modtagelse af nye bytes og begynder at behandle den, når vi får nok data.

Vi brugte bevidst ikke en stringValue - afkodning på en sådan måde ville være unødvendigt kompliceret. Derfor tilbyder Netty nyttige dekoderklasser, som er implementeringer af ChannelInboundHandler: ByteToMessageDecoder og Afspilning af dekoder.

Som vi bemærkede ovenfor, kan vi oprette en kanalbehandlingsrørledning med Netty. Så vi kan sætte vores dekoder som den første behandler, og behandlingslogikhåndtereren kan komme efter den.

Dekoderen til RequestData vises derefter:

offentlig klasse RequestDecoder udvider ReplayingDecoder {private final Charset charset = Charset.forName ("UTF-8"); @ Override beskyttet ugyldig dekodning (ChannelHandlerContext ctx, ByteBuf in, List out) kaster Undtagelse {RequestData data = nye RequestData (); data.setIntValue (in.readInt ()); int strLen = in.readInt (); data.setStringValue (in.readCharSequence (strLen, charset) .toString ()); out.add (data); }}

En idé om denne dekoder er ret enkel. Det bruger en implementering af ByteBuf der kaster en undtagelse, når der ikke er nok data i bufferen til læsning.

Når undtagelsen er fanget, spoles bufferen tilbage til starten, og dekoderen venter på en ny del af data. Afkodning stopper, når ud listen er ikke tom efter afkode udførelse.

3.4. Svarskoder

Udover afkodning af RequestData vi er nødt til at kode meddelelsen. Denne handling er enklere, fordi vi har de fulde meddelelsesdata, når skrivehandlingen sker.

Vi kan skrive data til Kanal i vores hovedhåndterer, eller vi kan adskille logikken og oprette en håndteringsudvidelse MessageToByteEncoder som får fat i skrivningen Svardata operation:

offentlig klasse ResponseDataEncoder udvider MessageToByteEncoder {@Override beskyttet ugyldig kode (ChannelHandlerContext ctx, ResponseData msg, ByteBuf ud) kaster Exception {out.writeInt (msg.getIntValue ()); }}

3.5. Behandling af anmodning

Da vi udførte afkodningen og kodningen i separate håndterere, er vi nødt til at ændre vores ProcessingHandler:

public class ProcessingHandler udvider ChannelInboundHandlerAdapter {@ Override public void channelRead (ChannelHandlerContext ctx, Object msg) kaster Undtagelse {RequestData requestData = (RequestData) msg; ResponseData responseData = nye ResponseData (); responseData.setIntValue (requestData.getIntValue () * 2); ChannelFuture fremtid = ctx.writeAndFlush (responsData); future.addListener (ChannelFutureListener.CLOSE); System.out.println (anmodningsdata); }}

3.6. Server Bootstrap

Lad os nu sætte det hele sammen og køre vores server:

offentlig klasse NettyServer {privat int port; // konstruktør offentlig statisk tomrum hoved (String [] args) kaster Undtagelse {int port = args.length> 0? Integer.parseInt (args [0]); : 8080; ny NettyServer (port) .run (); } public void run () kaster Undtagelse {EventLoopGroup bossGroup = ny NioEventLoopGroup (); EventLoopGroup workerGroup = ny NioEventLoopGroup (); prøv {ServerBootstrap b = ny ServerBootstrap (); b.group (bossGroup, workerGroup) .channel (NioServerSocketChannel.class) .childHandler (ny ChannelInitializer () {@Override public void initChannel (SocketChannel ch) kaster undtagelse {ch.pipeline (). addLast (ny RequestDecoder (), ny ResponseDataEc (), ny ProcessingHandler ());}}). valgmulighed (ChannelOption.SO_BACKLOG, 128) .childOption (ChannelOption.SO_KEEPALIVE, true); ChannelFuture f = b.bind (port) .sync (); f.channel (). closeFuture (). sync (); } endelig {workerGroup.shutdownGracefully (); bossGroup.shutdownGracefully (); }}}

Detaljerne om de klasser, der er brugt i ovenstående server bootstrap-eksempel, kan findes i deres Javadoc. Den mest interessante del er denne linje:

ch.pipeline (). addLast (ny RequestDecoder (), ny ResponseDataEncoder (), ny ProcessingHandler ());

Her definerer vi indgående og udgående håndtere, der behandler anmodninger og output i den rigtige rækkefølge.

4. Kundeapplikation

Klienten skal udføre omvendt kodning og afkodning, så vi skal have en RequestDataEncoder og ResponseDataDecoder:

offentlig klasse RequestDataEncoder udvider MessageToByteEncoder {private final Charset charset = Charset.forName ("UTF-8"); @ Override beskyttet ugyldig kode (ChannelHandlerContext ctx, RequestData msg, ByteBuf out) kaster Undtagelse {out.writeInt (msg.getIntValue ()); out.writeInt (msg.getStringValue (). længde ()); out.writeCharSequence (msg.getStringValue (), tegnsæt); }}

offentlig klasse ResponseDataDecoder udvider ReplayingDecoder {@ Override beskyttet ugyldig dekodning (ChannelHandlerContext ctx, ByteBuf in, List out) kaster Undtagelse {ResponseData data = nye ResponseData (); data.setIntValue (in.readInt ()); out.add (data); }}

Vi skal også definere en ClientHandler som sender anmodningen og modtager svaret fra serveren:

public class ClientHandler udvider ChannelInboundHandlerAdapter {@ Override public void channelActive (ChannelHandlerContext ctx) kaster Undtagelse {RequestData msg = new RequestData (); msg.setIntValue (123); msg.setStringValue ("alt arbejde og intet spil gør jack til en kedelig dreng"); ChannelFuture fremtid = ctx.writeAndFlush (msg); } @ Override public void channelRead (ChannelHandlerContext ctx, Object msg) kaster Undtagelse {System.out.println ((ResponseData) msg); ctx.close (); }}

Lad os nu bootstrap klienten:

offentlig klasse NettyClient {offentlig statisk ugyldig hoved (String [] args) kaster Undtagelse {String vært = "localhost"; int-port = 8080; EventLoopGroup workerGroup = ny NioEventLoopGroup (); prøv {Bootstrap b = ny Bootstrap (); b.group (workerGroup); b.kanal (NioSocketChannel.class); b.option (ChannelOption.SO_KEEPALIVE, sand); b.handler (ny ChannelInitializer () {@Override offentlig ugyldig initChannel (SocketChannel ch) kaster undtagelse {ch.pipeline (). addLast (ny RequestDataEncoder (), ny ResponseDataDecoder (), ny ClientHandler ());}}); ChannelFuture f = b.connect (vært, port) .sync (); f.kanal (). closeFuture (). synkronisering (); } endelig {workerGroup.shutdownGracefully (); }}}

Som vi kan se, er der mange detaljer til fælles med server bootstrapping.

Nu kan vi køre klientens hovedmetode og se på konsoludgangen. Som forventet fik vi det Svardata med intValue lig med 246.

5. Konklusion

I denne artikel havde vi en hurtig introduktion til Netty. Vi viste dets kernekomponenter som f.eks Kanal og ChannelHandler. Vi har også lavet en simpel ikke-blokerende protokolserver og en klient til den.

Som altid er alle kodeeksempler tilgængelige på GitHub.