Netty

傳統(tǒng)的IO模型的web容器，比如老版本的Tomcat，為了增加系統(tǒng)的吞吐量，需要不斷增加系統(tǒng)核心線(xiàn)程數(shù)量，或者通過(guò)水平擴(kuò)展服務(wù)器數(shù)量，來(lái)增加系統(tǒng)處理請(qǐng)求的能力。有了NIO之后，一個(gè)線(xiàn)程即可處理多個(gè)連接事件，其中基于多路復(fù)用模型的Netty框架，不僅降低了使用NIO的復(fù)雜度，

優(yōu)點(diǎn)

Netty是一款以java NIO為基礎(chǔ)，基于事件驅(qū)動(dòng)模型支持異步、高并發(fā)的網(wǎng)絡(luò)應(yīng)用框架。

• API使用簡(jiǎn)單，開(kāi)發(fā)門(mén)檻低，簡(jiǎn)化了NIO開(kāi)發(fā)網(wǎng)絡(luò)程序的復(fù)雜度
• 功能強(qiáng)大，預(yù)置多種編解碼功能，支持多種主流協(xié)議，比如Http、WebSocket。
• 定制能力強(qiáng)，可以通過(guò)ChannelHandler對(duì)通信框架靈活擴(kuò)展。
• 性能高，支持異步非阻塞通信模型
• 成熟穩(wěn)定，社區(qū)活躍，已經(jīng)修復(fù)了Java NIO所有的Bug。
• 經(jīng)歷了大規(guī)模商業(yè)應(yīng)用的考驗(yàn)，質(zhì)量有保證。

Reactor模型

Reactor模型也叫做反應(yīng)器設(shè)計(jì)模式，是一種為處理服務(wù)請(qǐng)求并發(fā)提交到一個(gè)或者多個(gè)服務(wù)處理器的事件設(shè)計(jì)模式。

而Netty是基于Reactor模型設(shè)計(jì)的一套高性能網(wǎng)絡(luò)應(yīng)用框架，下面來(lái)看下常見(jiàn)的Reactor模型有哪些：

一、單Reactor單線(xiàn)程

1）可以實(shí)現(xiàn)通過(guò)一個(gè)阻塞對(duì)象監(jiān)聽(tīng)多個(gè)鏈接請(qǐng)求

2）Reactor對(duì)象通過(guò)select監(jiān)聽(tīng)客戶(hù)端請(qǐng)求事件，通過(guò)dispatch進(jìn)行分發(fā)

3）如果是建立鏈接請(qǐng)求，則由Acceptor通過(guò)accept處理鏈接請(qǐng)求，然后創(chuàng)建一個(gè)Handler對(duì)象處理完成鏈接后的各種事件

4）如果不是鏈接請(qǐng)求，則由Reactor分發(fā)調(diào)用鏈接對(duì)應(yīng)的Handler來(lái)處理

5）Handler會(huì)完成Read->業(yè)務(wù)處理->send的完整業(yè)務(wù)流程

二、單Reactor多線(xiàn)程

1）Reactor對(duì)象通過(guò)select監(jiān)聽(tīng)客戶(hù)端請(qǐng)求事件，收到事件后，通過(guò)dispatch分發(fā)

2）如果是建立鏈接請(qǐng)求，則由Acceptor通過(guò)accept處理鏈接請(qǐng)求，然后創(chuàng)建一個(gè)Handler對(duì)象處理完成鏈接后的各種事件

3）如果不是鏈接請(qǐng)求，則由Reactor分發(fā)調(diào)用鏈接對(duì)應(yīng)的Handler來(lái)處理

4）Handler只負(fù)責(zé)事件響應(yīng)不做具體業(yè)務(wù)處理

5）通過(guò)read讀取數(shù)據(jù)后，分發(fā)到worker線(xiàn)程池處理，處理完成后返回給Handler，Handler收到后，通過(guò)send將結(jié)果返回給client

三、主從Reactor多線(xiàn)程

1）Reactor主線(xiàn)程MainReactor對(duì)象通過(guò)select監(jiān)聽(tīng)鏈接事件，通過(guò)Acceptor處理

2）當(dāng)Acceptor處理鏈接事件后，MainReactor將鏈接分配給SubReactor

3）SubReactor將鏈接加入到隊(duì)列進(jìn)行監(jiān)聽(tīng)，并創(chuàng)建Handler進(jìn)行事件處理

4）當(dāng)有新事件發(fā)生時(shí)，SubReactor就會(huì)調(diào)用對(duì)應(yīng)的Handler處理

5）Handler通過(guò)read讀取數(shù)據(jù)，分發(fā)到worker線(xiàn)程池處理，處理完成后返回給Handler，Handler收到后，通過(guò)send將結(jié)果返回給client

6）Reactor主線(xiàn)程可以對(duì)應(yīng)多個(gè)Reactor子線(xiàn)程

三種模式用生活案例來(lái)理解：

1）單Reactor單線(xiàn)程，前臺(tái)接待員和服務(wù)員是同一個(gè)人，全程為顧客服務(wù)

2）單Reactor多線(xiàn)程，1個(gè)前臺(tái)接待員，多個(gè)服務(wù)員，接待員只負(fù)責(zé)接待

3）主從Reactor多線(xiàn)程，多個(gè)前臺(tái)接待員，多個(gè)服務(wù)員

Reactor模型具有如下優(yōu)點(diǎn)：

1）響應(yīng)快，不必為單個(gè)同步事件所阻塞，雖然Reactor本身依然是同步的

2）可以最大程度的避免復(fù)雜的多線(xiàn)程及同步問(wèn)題，并且避免了多線(xiàn)程/進(jìn)程的切換開(kāi)銷(xiāo)

3）擴(kuò)展性好，可以方便的通過(guò)增加Reactor實(shí)例個(gè)數(shù)來(lái)充分利用CPU資源

4）復(fù)用性好，Reactor模型本身與具體事件處理邏輯無(wú)關(guān)，具有很高的復(fù)用性

核心組件

Bootstrap
一個(gè)Netty應(yīng)用通常由一個(gè)Bootstrap開(kāi)始，它主要作用是配置整個(gè)Netty程序，串聯(lián)起各個(gè)組件。

Handler
為了支持各種協(xié)議和處理數(shù)據(jù)的方式，便誕生了Handler組件。Handler主要用來(lái)處理各種事件，這里的事件很廣泛，比如可以是連接、數(shù)據(jù)接收、異常、數(shù)據(jù)轉(zhuǎn)換等。

ChannelInboundHandler
一個(gè)最常用的Handler。這個(gè)Handler的作用就是處理接收到數(shù)據(jù)時(shí)的事件，也就是說(shuō)，我們的業(yè)務(wù)邏輯一般就是寫(xiě)在這個(gè)Handler里面的，ChannelInboundHandler就是用來(lái)處理我們的核心業(yè)務(wù)邏輯。

ChannelInitializer
當(dāng)一個(gè)鏈接建立時(shí)，我們需要知道怎么來(lái)接收或者發(fā)送數(shù)據(jù)，當(dāng)然，我們有各種各樣的Handler實(shí)現(xiàn)來(lái)處理它，那么ChannelInitializer便是用來(lái)配置這些Handler，它會(huì)提供一個(gè)ChannelPipeline，并把Handler加入到ChannelPipeline。

ChannelPipeline
一個(gè)Netty應(yīng)用基于ChannelPipeline機(jī)制，這種機(jī)制需要依賴(lài)于EventLoop和EventLoopGroup，因?yàn)樗鼈內(nèi)齻€(gè)都和事件或者事件處理相關(guān)。

EventLoops
目的是為Channel處理IO操作，一個(gè)EventLoop可以為多個(gè)Channel服務(wù)。EventLoopGroup由多個(gè)EventLoop組成。

Channel
代表了一個(gè)Socket鏈接，或者其它和IO操作相關(guān)的組件，它和EventLoop一起用來(lái)參與IO處理。

Future
在Netty中所有的IO操作都是異步的，因此，你不能立刻得知消息是否被正確處理，但是我們可以過(guò)一會(huì)等它執(zhí)行完成或者直接注冊(cè)一個(gè)監(jiān)聽(tīng)，具體的實(shí)現(xiàn)就是通過(guò)Future和ChannelFutures,他們可以注冊(cè)一個(gè)監(jiān)聽(tīng)，當(dāng)操作執(zhí)行成功或失敗時(shí)監(jiān)聽(tīng)會(huì)自動(dòng)觸發(fā)。

示例

下面通過(guò)一個(gè)簡(jiǎn)單的示例，了解怎么基于Netty如何開(kāi)發(fā)一個(gè)簡(jiǎn)單的網(wǎng)絡(luò)通信程序。和之前一樣，包含服務(wù)端與客戶(hù)端：

Server:

@Slf4j
public class Server {

    private EventLoopGroup boosGroup;

    private EventLoopGroup workGroup;

    public Server(int port){
        try {
            init(port);
            log.info("----- 服務(wù)啟動(dòng)成功 -----");
        } catch (InterruptedException e) {
            log.error("啟動(dòng)服務(wù)出錯(cuò)：{}", e.getCause());
        }
    }

    private void init(int port) throws InterruptedException {
        // 處理連接
        this.boosGroup = new NioEventLoopGroup();
        // 處理業(yè)務(wù)
        this.workGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        // 綁定
        bootstrap.group(boosGroup, workGroup)
                .channel(NioServerSocketChannel.class) //配置服務(wù)端
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_RCVBUF, 1024)
                .childOption(ChannelOption.SO_SNDBUF, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new ServerHandler());
                    }
                });

        ChannelFuture channelFuture = bootstrap.bind(port).sync();
        channelFuture.channel().closeFuture().sync();
    }

    public void close(){
        this.boosGroup.shutdownGracefully();
        this.workGroup.shutdownGracefully();
    }

}

@Slf4j
class ServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info(">>>>>>> server active");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //1. 讀取客戶(hù)端的數(shù)據(jù)(緩存中去取并打印到控制臺(tái))
        ByteBuf buf = (ByteBuf) msg;
        byte[] request = new byte[buf.readableBytes()];
        buf.readBytes(request);
        String requestBody = new String(request, "utf-8");
        log.info(">>>>>>>>> receive message: {}", requestBody);

        //2. 返回響應(yīng)數(shù)據(jù)
        ctx.writeAndFlush(Unpooled.copiedBuffer((requestBody+" too").getBytes()));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

Client:

@Slf4j
public class Client {

    private EventLoopGroup workGroup;
    private ChannelFuture channelFuture;

    public Client(int port){
        init(port);
    }

    private void init(int port){
        this.workGroup = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .option(ChannelOption.SO_RCVBUF, 1024)
                .option(ChannelOption.SO_SNDBUF, 1024)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new ClientHandler());
                    }
                });

        this.channelFuture = bootstrap.connect("127.0.0.1", port).syncUninterruptibly();
    }

    /**
     *
     * @param message
     */
    public void send(String message){
        this.channelFuture.channel().writeAndFlush(Unpooled.copiedBuffer(message.getBytes()));
    }

    /**
     *
     */
    public void close(){
        try {
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        workGroup.shutdownGracefully();
    }
}

@Slf4j
class ClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info(">>>>>>> client active");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);

            String body = new String(req, "utf-8");
            log.info(">>>>>>>>> receive message: {}", body);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

測(cè)試：

public class StarterTests {

    static int port = 9011;

    @Test
    public void startServer(){
        Server server = new Server(9011);
    }

    @Test
    public void startClient(){
        Client client = new Client(port);
        client.send("Hello Netty!");
        while (true){}
    }

}

主要實(shí)現(xiàn)步驟：

1、創(chuàng)建兩個(gè)NIO線(xiàn)程組，一個(gè)專(zhuān)門(mén)用來(lái)網(wǎng)絡(luò)事件處理（接受客戶(hù)端連接），另一個(gè)則進(jìn)行網(wǎng)絡(luò)通訊讀寫(xiě)

2、創(chuàng)建一個(gè)ServerBootstrap對(duì)象，配置Netty的一系列參數(shù)，例如接收傳入數(shù)據(jù)的緩存大小等。

3、創(chuàng)建一個(gè)實(shí)際處理數(shù)據(jù)的類(lèi)ChannelInitializer,進(jìn)行初始化的準(zhǔn)備工作，比如設(shè)置傳入數(shù)據(jù)的字符集格式，實(shí)現(xiàn)具體處理數(shù)據(jù)的接口。

4、綁定端口，執(zhí)行同步阻塞方法等待服務(wù)器啟動(dòng)即可

類(lèi)似技術(shù)

Mina、Netty、Grizzly

1、Mina

Mina(Multipurpose Infrastructure for Network Applications) 是 Apache 組織一個(gè)較新的項(xiàng)目，它為開(kāi)發(fā)高性能和高可用性的網(wǎng)絡(luò)應(yīng)用程序提供了非常便利的框架。當(dāng)前發(fā)行的 Mina 版本2.04支持基于 Java NIO 技術(shù)的 TCP/UDP 應(yīng)用程序開(kāi)發(fā)、串口通訊程序，Mina 所支持的功能也在進(jìn)一步的擴(kuò)展中。

目前，正在使用Mina的應(yīng)用包括：Apache Directory Project、AsyncWeb、AMQP（Advanced Message Queuing Protocol）、RED5 Server（Macromedia Flash Media RTMP）、ObjectRADIUS、 Openfire等等。

2、Netty

Netty是一款異步的事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用框架和工具，用于快速開(kāi)發(fā)可維護(hù)的高性能、高擴(kuò)展性協(xié)議服務(wù)器和客戶(hù)端。也就是說(shuō)，Netty是一個(gè)NIO客戶(hù)端/服務(wù)器框架，支持快速、簡(jiǎn)單地開(kāi)發(fā)網(wǎng)絡(luò)應(yīng)用，如協(xié)議服務(wù)器和客戶(hù)端。它極大簡(jiǎn)化了網(wǎng)絡(luò)編程，如TCP和UDP套接字服務(wù)器。

3、Grizzly

Grizzly是一種應(yīng)用程序框架，專(zhuān)門(mén)解決編寫(xiě)成千上萬(wàn)用戶(hù)訪問(wèn)服務(wù)器時(shí)候產(chǎn)生的各種問(wèn)題。使用JAVA NIO作為基礎(chǔ)，并隱藏其編程的復(fù)雜性。容易使用的高性能的API。帶來(lái)非阻塞socketd到協(xié)議處理層。利用高性能的緩沖和緩沖管理使用高性能的線(xiàn)程池。

從設(shè)計(jì)的理念上來(lái)看，Mina的設(shè)計(jì)理念是最為優(yōu)雅的。當(dāng)然，由于Netty的主導(dǎo)作者與Mina的主導(dǎo)作者是同一人，出自同一人之手的Netty在設(shè)計(jì)理念上與Mina基本上是一致的。而Grizzly在設(shè)計(jì)理念上就較差了點(diǎn)，幾乎是JavaNIO的簡(jiǎn)單封裝。

結(jié)束語(yǔ)

前面我們說(shuō)到了各種IO模型，基于Java NIO開(kāi)發(fā)的網(wǎng)絡(luò)應(yīng)用程序，能夠更好的基于服務(wù)器操作系統(tǒng)底層支持，更好的利用服務(wù)器資源提供高性能、可靠易擴(kuò)展的應(yīng)用。而Netty在Java NIO的基礎(chǔ)上，為開(kāi)發(fā)這提供便捷提升開(kāi)發(fā)的效率與穩(wěn)定性，在很多主流的框架中都有被使用，比如Dubbo、Spring Webflux等