Preskoči na glavni sadržaj

Veoma detaljan uvod u Netty, dovoljno je pročitati ovaj članak!

Chenmo Wang ErSegmentfaultnetty3390 reči11 minuta

Predgovor

Ovaj članak uglavnom govori o osobinama i važnim komponentama Netty okvira, uz nadu da ćete nakon čitanja steći prilično intuitivan osećaj za Netty okvir, da će čitaocima pomoći da brzo uđu u Netty i izbegnu neke zamke.

I. Pregled Netty-ja

Službeni opis:

Netty is an asynchronous event-driven network application framework

for rapid development of maintainable high performance protocol servers & clients.

Netty je asinhroni, događajima vođeni okvir za mrežne aplikacije, namenjen brzom razvoju održivih servera i klijenata visokih performansi za protokole.

II. Zašto koristiti Netty

Prema prezentaciji na službenoj stranici, Netty je okvir za mrežne aplikacije kojim se razvijaju serveri i klijenti — dakle okvir za mrežno programiranje. Pošto je u pitanju mrežno programiranje, o Socket-u nećemo ni govoriti; zašto onda ne iskoristiti NIO?

2.1 Mane NIO-a

Za to pitanje sam ranije napisao članak Uvod u NIO sa prilično detaljnim uvodom. Glavni problemi NIO-a su:

  • Biblioteke klasa i API NIO-a su složeni, a cena učenja visoka — morate dobro savladati Selector, ServerSocketChannel, SocketChannel, ByteBuffer itd.
  • Potrebno je dobro poznavati višestruko programiranje u Javi. Razlog je što NIO programiranje uključuje Reactor obrazac, pa morate biti veoma vešti u višestrukom i mrežnom programiranju da biste napisali kvalitetne NIO programe.
  • Notorni epoll bug. On dovodi do praznog poliranja (busy-poll) Selector-a, što na kraju rezultira CPU od 100%. Čak ni u JDK 1.7 ovo nije fundamentalno rešeno.

2.2 Prednosti Netty-ja

S druge strane, Netty ima mnogo prednosti:

  • API je jednostavan za upotrebu, a cena učenja niska.
  • Snažne funkcije — ugrađeni su razni dekoderi i enkoderi, uz podršku za više protokola.
  • Visoke performanse — u poređenju sa drugim glavnim NIO okvirima, Netty ima najbolje performanse.
  • Aktivna zajednica — greške se brzo ispravljaju, ciklus iteracija verzija je kratak, a nove funkcije se stalno dodaju.
  • I Dubbo i Elasticsearch koriste Netty, što potvrđuje njegov kvalitet.

III. Dijagram arhitekture

Ova slika iznad je dijagram arhitekture sa naslovne stranice službenog sajta; analizirajmo ga odozgo nadole.

Zeleni deo je Core jezgroviti modul, koji obuhvata zero-copy, API biblioteku i proširivi model događaja.

Narandžasti deo je Protocol Support — podrška za protokole, uključujući Http protokol, WebSocket, SSL (Secure Sockets Layer), Google Protobuf protokol, zlib/gzip kompresiju i dekompresiju, Large File Transfer (prenos velikih datoteka) itd.

Crveni deo je Transport Services — servisi prenosa, uključujući Socket, Datagram, Http Tunnel itd.

Iz gore navedenog vidi se da su funkcije, protokoli i načini prenosa Netty-ja prilično sveobuhvatni i snažni.

IV. Večiti Hello World

Prvo ćemo napraviti jedan HelloWorld projekat, da bismo se upoznali sa API-jem i pripremili teren za kasnije učenje. Polazimo od sledeće slike:

4.1 Dodavanje Maven zavisnosti

Verzija koju koristimo je 4.1.20, što je relativno stabilna verzija.

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.20.Final</version>
</dependency>

4.2 Kreiranje klase za pokretanje servera

public class MyServer {
    public static void main(String[] args) throws Exception {
        // Kreiramo dve grupe niti: bossGroup i workerGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            // Kreiramo objekat za pokretanje servera i podesimo parametre
            ServerBootstrap bootstrap = new ServerBootstrap();
            // Podesimo dve grupe niti: bossGroup i workerGroup
            bootstrap.group(bossGroup, workerGroup)
                // Podesimo tip implementacije kanala servera
                .channel(NioServerSocketChannel.class)
                // Podesimo broj veza u redu čekanja niti
                .option(ChannelOption.SO_BACKLOG, 128)
                // Podesimo održavanje aktivne veze
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                // Inicijalizujemo objekat kanala pomoću anonimne unutrašnje klase
                .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // Postavimo procesor na pipeline
                            socketChannel.pipeline().addLast(new MyServerHandler());
                        }
                    });// Postavimo procesor na kanal koji odgovara EventLoop-u iz workerGroup
            System.out.println("Server ljubitelja Java tehnologije je spreman...");
            // Vežemo broj porta i pokrenemo server
            ChannelFuture channelFuture = bootstrap.bind(6666).sync();
            // Slušamo zatvaranje kanala
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

4.3 Kreiranje procesora servera

/**
 * Prilagođeni Handler mora nasleđivati HandlerAdapter koji je propisao Netty,
 * kako bi ga Netty okvir povezao — donekle podseća na adapter obrazac iz SpringMVC.
 **/
public class MyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // Preuzmemo poruku koju je poslao klijent
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("Primljena poruka od klijenta " + ctx.channel().remoteAddress() + ": " + byteBuf.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // Pošaljemo poruku klijentu
        ctx.writeAndFlush(Unpooled.copiedBuffer("Server je primio poruku i šalje ti pitanje?", CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // Pri pojavi izuzetka zatvorimo kanal
        ctx.close();
    }
}

4.4 Kreiranje klase za pokretanje klijenta

public class MyClient {

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            // Kreiramo bootstrap objekat i podesimo parametre
            Bootstrap bootstrap = new Bootstrap();
            // Podesimo grupu niti
            bootstrap.group(eventExecutors)
                // Podesimo tip implementacije kanala klijenta
                .channel(NioSocketChannel.class)
                // Inicijalizujemo kanal pomoću anonimne unutrašnje klase
                .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // Dodamo procesor kanala klijenta
                            ch.pipeline().addLast(new MyClientHandler());
                        }
                    });
            System.out.println("Klijent je spreman, može da poleće~");
            // Povežemo se na server
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
            // Slušamo zatvaranje kanala
            channelFuture.channel().closeFuture().sync();
        } finally {
            // Zatvorimo grupu niti
            eventExecutors.shutdownGracefully();
        }
    }
}

4.5 Kreiranje procesora klijenta

public class MyClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // Pošaljemo poruku serveru
        ctx.writeAndFlush(Unpooled.copiedBuffer("Waibibabu~ Jasmine~ Are you good~ Malezija~", CharsetUtil.UTF_8));
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // Primimo poruku koju je poslao server
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("Primljena poruka od servera " + ctx.channel().remoteAddress() + ": " + byteBuf.toString(CharsetUtil.UTF_8));
    }
}

4.6 Testiranje

Prvo pokrenite server, a zatim klijent — i videćete rezultat:

Ispis MyServer-a:

Ispis MyClient-a:

V. Osobine Netty-ja i važne komponente

5.1 Red zadataka taskQueue

Ako Handler ima neke dugotrajne poslovne obrade, možemo ih predati taskQueue na asinhronu obradu. Kako se to radi? Pogledajte demonstraciju koda:

public class MyServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // Preuzmemo pool niti eventLoop, dodamo nit i izvršimo
        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    // Dugotrajna operacija — neće dozvoliti da dugotrajna poslovna obrada blokira Handler
                    Thread.sleep(1000);
                    System.out.println("Dugotrajna poslovna obrada");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

Ako postavimo jedno debug zaustavljanje, možemo videti da dodati taskQueue sadrži jedan zadatak.

5.2 Red odloženih zadataka scheduleTaskQueue

Red odloženih zadataka je vrlo sličan prethodno opisanom redu zadataka, jedina razlika je u tome što dodaje mogućnost odlaganja izvršenja za određeno vreme. Pogledajte demonstraciju koda:

ctx.channel().eventLoop().schedule(new Runnable() {
    @Override
    public void run() {
        try {
            // Dugotrajna operacija — neće dozvoliti da dugotrajna poslovna obrada blokira Handler
            Thread.sleep(1000);
            System.out.println("Dugotrajna poslovna obrada");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
},5, TimeUnit.SECONDS);// Izvršava se nakon 5 sekundi

I dalje u debug režimu možemo videti da postoji jedan scheduleTaskQueue zadatak na čekanju izvršenja.

5.3 Asinhroni mehanizam Future

Prilikom izrade HelloWorld projekta videli smo jedan ovakav red koda:

ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666);

Mnoge operacije vraćaju taj ChannelFuture objekat — za šta on zapravo služi?

ChannelFuture pruža način asinhronog obaveštavanja po završetku operacije. U Socket programiranju se čekanje rezultata obično obavlja sinhrono, blokirajući, dok Netty ne izaziva blokiranje — jer ChannelFuture rezultate preuzima na način sličan observer obrascu. Pogledajte kratku demonstraciju koda:

// Dodamo slušaoca
channelFuture.addListener(new ChannelFutureListener() {
    // Anonimna unutrašnja klasa ChannelFutureListener interfejsa
    // Prepisujemo metod operationComplete
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        // Proverimo da li je operacija uspela
        if (future.isSuccess()) {
            System.out.println("Povezivanje uspešno");
        } else {
            System.out.println("Povezivanje neuspešno");
        }
    }
});

5.4 Bootstrap i ServerBootStrap

Bootstrap i ServerBootStrap su fabričke klase koje Netty pruža za kreiranje pokretača klijenta i servera. Korišćenjem ove fabričke klase pokretačke klase se kreiraju veoma pogodno — prema prethodnim primerima, vidi se i da značajno smanjuje težinu razvoja. Najpre pogledajte jedan dijagram klasa:

Vidi se da obe nasleđuju apstraktnu klasu AbstractBootStrap, pa su konfiguracioni metodi uglavnom isti.

Uglavnom, koraci za kreiranje pokretača pomoću Bootstrap-a mogu se podeliti na sledeći način:

5.4.1 group()

U prethodnom članku Reactor obrazac već smo objasnili da server treba da koristi dve grupe niti:

  • bossGroup služi za slušanje klijentskih veza, isključivo je zadužen za uspostavljanje veze sa klijentom i registraciju te veze u Selector workerGroup-e.
  • workerGroup služi za obradu događaja čitanja i pisanja koji se javljaju na svakoj vezi.

Pri kreiranju grupa niti obično je dovoljno uraditi sledeće new:

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();

Zanima me jedno: pošto su u pitanju grupe niti, koliki je podrazumevani broj niti? Zaronimo u izvorni kod:

// Sačuvano u jednoj konstanti
    private static final int DEFAULT_EVENT_LOOP_THREADS;

    static {
        // NettyRuntime.availableProcessors() * 2 — dvostruka vrednost broja jezgara CPU-a dodeljuje se konstanti
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

        if (logger.isDebugEnabled()) {
            logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
        }
    }
    
    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        // Ako se ne prosledi, koristi se vrednost konstante, dakle dvostruka vrednost broja jezgara CPU-a
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

Iz izvornog koda se vidi da je podrazumevani broj niti dvostruka vrednost broja jezgara CPU-a. Ako želite da prilagodite broj niti, možete koristiti parametrizovani konstruktor:

// Podesi broj niti bossGroup-a na 1
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// Podesi broj niti workerGroup-a na 16
EventLoopGroup workerGroup = new NioEventLoopGroup(16);

5.4.2 channel()

Ovaj metod služi za postavljanje tipa kanala; nakon uspostavljanja veze, na osnovu ovog podešavanja kreira se odgovarajuća instanca Channel-a.

U debug režimu možemo videti:

Tipovi kanala su sledeći:

NioSocketChannel: asinhrona, neblokirajuća klijentska TCP Socket veza.

NioServerSocketChannel: asinhrona, neblokirajuća serverska TCP Socket veza.

Najčešće se koriste ova dva tipa kanala, jer su asinhrona i neblokirajuća, pa su prvi izbor.

OioSocketChannel: sinhrona, blokirajuća klijentska TCP Socket veza.

OioServerSocketChannel: sinhrona, blokirajuća serverska TCP Socket veza.

Malo sam probao lokalno; ponaša se drugačije od Nio-a, jer blokira, pa je i API drugačiji. Pošto je u pitanju blokirajući IO, gotovo da niko ne bira Oio, pa je teško pronaći primere. Malo sam se potrudio i, nakon nekoliko grešaka, uspeo da ga pokrenem. Kod je sledeći:

// Serverski kod, skoro identičan kao gore, potrebno je izmeniti samo tri mesta
// Ovde se koristi OioEventLoopGroup
EventLoopGroup bossGroup = new OioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup)// Potrebno je podesiti samo jednu grupu niti bossGroup
        .channel(OioServerSocketChannel.class)// Podesi tip implementacije kanala servera

// Klijentski kod, potrebno je izmeniti samo dva mesta
// Koristi se OioEventLoopGroup
EventLoopGroup eventExecutors = new OioEventLoopGroup();
// Tip kanala se postavlja na OioSocketChannel
bootstrap.group(eventExecutors)// Podesi grupu niti
        .channel(OioSocketChannel.class)// Podesi tip implementacije kanala klijenta

NioSctpChannel: asinhrona klijentska Sctp (Stream Control Transmission Protocol, protokol za kontrolu prenosa toka) veza.

NioSctpServerChannel: asinhrona serverska Sctp veza.

Nisam uspeo da ga pokrenem lokalno; pregledao sam komentare korisnika na mreži i kažu da se može pokrenuti samo u Linux okruženju. Iz poruke o grešci: "SCTP not supported on this platform" — platforma nije podržana. Pošto je moj računar Windows sistem, čini se da korisnici imaju pravo.

5.4.3 option() i childOption()

Prvo da pojasnimo razliku između ova dva.

option() podešava parametre za serversku stranu koja prima dolazne veze, dakle za bossGroup nit.

childOption() se odnosi na veze koje roditeljski kanal primi, dakle za workerGroup nit.

Kada smo to razjasnili, pogledajmo koje se postavke najčešće koriste:

Parametri za SocketChannel, dakle uobičajeni parametri za childOption():

SO_RCVBUF — Socket parametar, veličina bafera za prijem TCP podataka.

TCP_NODELAY — TCP parametar, šalje podatke odmah; podrazumevana vrednost je True.

SO_KEEPALIVE — Socket parametar, održavanje veze aktivnom; podrazumevana vrednost je False. Kada je omogućen, TCP aktivno proverava ispravnost neaktivnih veza.

Parametri za ServerSocketChannel, dakle uobičajeni parametri za option():

SO_BACKLOG — Socket parametar, dužina reda veza koje server prihvata; ako je red popunjen, klijentske veze biće odbijene. Podrazumevana vrednost na Windows-u je 200, a na ostalim sistemima 128.

Zbog ograničenja prostora, ostale nećemo navoditi — možete potražiti materijale na mreži i upoznati se sa njima.

5.4.4 Postavljanje pipeline-a (ključno)

ChannelPipeline je lanac odgovornosti kojim Netty obrađuje zahteve, dok je ChannelHandler konkretan procesor za obradu zahteva. Zapravo, svaki channel ima svoj pipeline procesora.

U Bootstrap-u, metod childHandler() treba da inicijalizuje kanal, odnosno da instancira jedan ChannelInitializer — tada je potrebno prepisati metod initChannel() kojim se inicijalizuje kanal; tu se obavlja i sklapanje pipeline-a. Demonstracija koda:

// Inicijalizujemo objekat kanala pomoću anonimne unutrašnje klase
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        // Na pipeline postavimo prilagođeni procesor
        socketChannel.pipeline().addLast(new MyServerHandler());
    }
});

Procesori Handler se uglavnom dele na dve vrste:

ChannelInboundHandlerAdapter (inbound procesor), ChannelOutboundHandler (outbound procesor)

Inbound znači da podaci idu sa osnovnog java NIO Channel-a u Netty Channel.

Outbound znači da se osnovni java NIO Channel operiše preko Netty Channel-a.

Česti događaji kod ChannelInboundHandlerAdapter procesora:

  1. Događaj registracije fireChannelRegistered.
  2. Događaj uspostavljanja veze fireChannelActive.
  3. Događaj čitanja i događaj završetka čitanja fireChannelRead, fireChannelReadComplete.
  4. Događaj obaveštavanja o izuzetku fireExceptionCaught.
  5. Korisnički definisan događaj fireUserEventTriggered.
  6. Događaj promene stanja upisivosti Channel-a fireChannelWritabilityChanged.
  7. Događaj zatvaranja veze fireChannelInactive.

Česti događaji kod ChannelOutboundHandler procesora:

  1. Bind — vezivanje porta.
  2. Connect — povezivanje na server.
  3. Write — događaj pisanja.
  4. Flush — događaj ispiranja.
  5. Read — događaj čitanja.
  6. Disconnect — aktivno prekidanje veze.
  7. Close — događaj zatvaranja kanala.

Postoji i sličan metod handler(), koji se uglavnom koristi za sklapanje parent kanala, dakle bossGroup niti. U opštem slučaju se ovaj metod ne koristi.

5.4.5 bind()

Služi serveru ili klijentu da veže adresu servera i broj porta; podrazumevano se pokreće asinhrono. Ako se doda metod sync(), onda je sinhrono.

Postoji pet istoimenih preopterećenih metoda, čija je svrha vezivanje adrese i porta. Nećemo ih pojedinačno objašnjavati.

5.4.6 Eleganntno zatvaranje EventLoopGroup-a

// Oslobađamo sve resurse, uključujući kreirane niti
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();

Time se zatvaraju svi child kanali. Nakon zatvaranja, oslobađaju se osnovni resursi.

5.5 Channel

Šta je Channel? Pogledajmo objašnjenje u službenoj dokumentaciji:

A nexus to a network socket or a component which is capable of I/O operations such as read, write, connect, and bind

Prevod ukratko: komponenta koja povezuje mrežni socket ili koja je sposobna za I/O operacije poput čitanja, pisanja, povezivanja i vezivanja.

Ako vam gornje objašnjenje deluje apstraktno, evo još jednog:

A channel provides a user:

the current state of the channel (e.g. is it open? is it connected?),

the configuration parameters of the channel (e.g. receive buffer size),

the I/O operations that the channel supports (e.g. read, write, connect, and bind), and

the ChannelPipeline which handles all I/O events and requests associated with the channel.

Prevod ukratko:

channel korisniku pruža:

  1. Trenutno stanje kanala (npr. da li je otvoren? da li je povezan?)
  2. Konfiguracione parametre kanala (npr. veličinu bafera za prijem)
  3. I/O operacije koje kanal podržava (npr. čitanje, pisanje, povezivanje i vezivanje), kao i ChannelPipeline koji obrađuje sve I/O događaje i zahteve povezane sa kanalom.

5.5.1 Preuzimanje stanja kanala

boolean isOpen(); // Ako je kanal otvoren, vraća true
boolean isRegistered();// Ako je kanal registrovan u EventLoop, vraća true
boolean isActive();// Ako je kanal u aktivnom stanju i povezan, vraća true
boolean isWritable();// Kada i samo kada će I/O nit odmah izvršiti traženu operaciju pisanja, vraća true.

To su četiri metoda za preuzimanje stanja kanala.

5.5.2 Preuzimanje konfiguracionih parametara kanala

Za preuzimanje jedne konfiguracione informacije koristi se getOption(). Demonstracija koda:

ChannelConfig config = channel.config();// Preuzmemo konfiguracione parametre
// Preuzmemo parametar ChannelOption.SO_BACKLOG
Integer soBackLogConfig = config.getOption(ChannelOption.SO_BACKLOG);
// Pošto sam u pokretaču podesio 128, ovde dobijam soBackLogConfig=128

Za preuzimanje više konfiguracionih informacija koristi se getOptions(). Demonstracija koda:

ChannelConfig config = channel.config();
Map<ChannelOption<?>, Object> options = config.getOptions();
for (Map.Entry<ChannelOption<?>, Object> entry : options.entrySet()) {
    System.out.println(entry.getKey() + " : " + entry.getValue());
}
/**
SO_REUSEADDR : false
WRITE_BUFFER_LOW_WATER_MARK : 32768
WRITE_BUFFER_WATER_MARK : WriteBufferWaterMark(low: 32768, high: 65536)
SO_BACKLOG : 128
ostalo izostavljeno...
*/

5.5.3 I/O operacije koje kanal podržava

Operacija pisanja — ovde demonstriramo slanje poruke sa servera ka klijentu:

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.channel().writeAndFlush(Unpooled.copiedBuffer("Ovo je, ovo je meso-jaje-luk-piletina~", CharsetUtil.UTF_8));
}

Klijentska konzola:

// Primljena poruka od servera /127.0.0.1:6666: Ovo je, ovo je meso-jaje-luk-piletina~

Operacija povezivanja, demonstracija koda:

ChannelFuture connect = channelFuture.channel().connect(new InetSocketAddress("127.0.0.1", 6666));// Obično se koristi pokretač; ovaj način se retko koristi

Preuzimanje ChannelPipeline-a preko channel-a, i srodna obrada:

// Preuzmemo objekat ChannelPipeline
ChannelPipeline pipeline = ctx.channel().pipeline();
// U pipeline dodamo ChannelHandler procesor — sklapanje pipeline-a
pipeline.addLast(new MyServerHandler());

5.6 Selector

U NioEventLoop-u postoji članska promenljiva selector — to je Selector iz nio paketa. U prethodnom članku Uvod u NIO već sam objasnio Selector.

Selector u Netty-ju je isti kao Selector u NIO-u — služi za slušanje događaja, upravljanje kanalima registrovanim u Selector-u i realizaciju multipleksera.

5.7 PiPeline i ChannelPipeline

Ranije, prilikom predstavljanja Channel-a, videli smo da se u channel-u može sklopiti ChannelPipeline procesor. Jedan channel ne može imati samo jedan channelHandler procesor — njih sigurno ima više. Pošto više channelHandler-a radi u jednom pipeline-u, nužno postoji redosled.

Zato se pojavljuje pipeline — on služi kao kontejner za procesore. Pri inicijalizaciji channel-a, channelHandler-i se po redosledu ubacuju u pipeline, čime se postiže izvršavanje channelHandler-a po redu.

U jednom Channel-u postoji samo jedan ChannelPipeline. Taj pipeline se kreira u trenutku kreiranja Channel-a. ChannelPipeline sadrži listu oblikovanu od ChannelHandler-a, i svi ChannelHandler-i se registruju u ChannelPipeline.

5.8 ChannelHandlerContext

U Netty-ju Handler procesor definišemo sami — ranije je rečeno da se to radi nasleđivanjem inbound ili outbound procesora. Ako tada unutar Handler-a želimo da preuzmemo pipeline objekat ili channel objekat — kako?

Zato je Netty dizajnirao ovaj ChannelHandlerContext kontekstni objekat, pomoću koga možemo doći do channel-a, pipeline-a i drugih objekata, i izvršavati operacije poput čitanja i pisanja.

Preko dijagrama klasa, ChannelHandlerContext je interfejs sa tri implementacione klase ispod.

Zapravo, ChannelHandlerContext u pipeline-u ima oblik povezane liste. Pogledajte jedan deo izvornog koda i biće vam jasno:

// Konstruktor metoda DefaultChannelPipeline, implementacione klase ChannelPipeline
protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);
    // Postavi head čvor head i tail čvor tail
    tail = new TailContext(this);
    head = new HeadContext(this);
    
    head.next = tail;
    tail.prev = head;
}

Evo slike koja to predstavlja, a koja je jasnija:

5.9 EventLoopGroup

Prvo pogledajmo dijagram klasa EventLoopGroup-a:

Među njima je i često korišćena implementaciona klasa NioEventLoopGroup. OioEventLoopGroup je takođe korišćena u prethodnim primerima.

Iz dijagrama Netty arhitekture može se videti da server za rad zahteva saradnju dve grupe niti, a interfejs te grupe niti je upravo EventLoopGroup.

Svaka EventLoopGroup sadrži jedan ili više EventLoop-a, a svaki EventLoop održava jednu instancu Selector-a.

5.9.1 Princip realizacije mehanizma round-robin-a

Pogledajmo jedan deo izvornog koda DefaultEventExecutorChooserFactory:

private final AtomicInteger idx = new AtomicInteger();
private final EventExecutor[] executors;

@Override
public EventExecutor next() {
    // idx.getAndIncrement() odgovara idx++, zatim se uzima po modulu dužine niza zadataka
    return executors[idx.getAndIncrement() & executors.length - 1];
}

Ovaj kod potvrđuje da je način izvršavanja mehanizam round-robin-a. Hajde da ga debug-ujemo:

Ovde postoji i jedna provera — ako broj niti nije stepen dvojke, koristi se algoritam po modulu.

@Override
public EventExecutor next() {
    return executors[Math.abs(idx.getAndIncrement() % executors.length)];
}

Link preuzeto sa: https://segmentfault.com/a/1190000023840032, priredio: Chenmo Wang Er