小六六學Netty系列之初識Netty

前言

文本已收錄至我的GitHub倉庫

種一棵樹最好的時間是十年前，其次是現在

我知道很多人不玩qq了,但是懷舊一下,歡迎加入六脈神劍Java菜鳥學習群，群聊号碼：549684836 鼓勵大家在技術的路上寫部落格

絮叨

為了學習Netty，我們前面鋪墊了那麼多，NIO Java的零拷貝，UNIX的I/O模型等等。

今天我們就來看看Netty 然後用Netty搞個最簡單的例子

Netty官網

Netty官網說明

Netty 是由 JBOSS 提供的一個 Java 開源架構。Netty 提供異步的、基于事件驅動的網絡應用程式架構，用以快速開發高性能、高可靠性的網絡 IO 程式
Netty 可以幫助你快速、簡單的開發出一個網絡應用，相當于簡化和流程化了 NIO 的開發過程
Netty 是目前最流行的 NIO 架構，Netty 在網際網路領域、大資料分布式計算領域、遊戲行業、通信行業等獲得了廣泛的應用，知名的 Elasticsearch 、Dubbo 架構内部都采用了 Netty。

Netty的優點

Netty 對 JDK 自帶的 NIO 的 API 進行了封裝，解決了上述問題。

設計優雅：适用于各種傳輸類型的統一 API 阻塞和非阻塞 Socket；基于靈活且可擴充的事件模型，可以清晰地分離關注點；高度可定制的線程模型 - 單線程，一個或多個線程池.
使用友善：詳細記錄的 Javadoc，使用者指南和示例；沒有其他依賴項，JDK 5（Netty 3.x）或 6（Netty 4.x）就足夠了。

高性能、吞吐量更高：延遲更低；減少資源消耗；最小化不必要的記憶體複制。

安全：完整的 SSL/TLS 和 StartTLS 支援。

社群活躍、不斷更新：社群活躍，版本疊代周期短，發現的 Bug 可以被及時修複，同時，更多的新功能會被加入

Reactor 模式

針對傳統阻塞 I/O 服務模型的 2 個缺點，解決方案：

基于 I/O 複用模型：多個連接配接共用一個阻塞對象，應用程式隻需要在一個阻塞對象等待，無需阻塞等待所有連接配接。當某個連接配接有新的資料可以處理時，作業系統通知應用程式，線程從阻塞狀态傳回，開始進行業務處理Reactor 對應的叫法: 1. 反應器模式 2. 分發者模式(Dispatcher) 3. 通知者模式(notifier)
基于線程池複用線程資源：不必再為每個連接配接建立線程，将連接配接完成後的業務處理任務配置設定給線程進行處理，一個線程可以處理多個連接配接的業務。

I/O 複用結合線程池，就是 Reactor 模式基本設計思想，如圖：

Reactor 模式，通過一個或多個輸入同時傳遞給服務處理器的模式(基于事件驅動)
伺服器端程式處理傳入的多個請求,并将它們同步分派到相應的處理線程，是以Reactor模式也叫 Dispatcher模式
Reactor 模式使用IO複用監聽事件, 收到事件後，分發給某個線程(程序), 這點就是網絡伺服器高并發處理關鍵

Reactor 三種模式

單 Reactor 單線程，前台接待員和服務員是同一個人，全程為顧客服
單 Reactor 多線程，1 個前台接待員，多個服務員，接待員隻負責接待
主從 Reactor 多線程，多個前台接待員，多個服務生

Netty模型

工作原理示意圖

Netty抽象出兩組線程池 BossGroup 專門負責接收用戶端的連接配接, WorkerGroup 專門負責網絡的讀寫

BossGroup 和 WorkerGroup 類型都是 NioEventLoopGroup

NioEventLoopGroup 相當于一個事件循環組, 這個組中含有多個事件循環，每一個事件循環是 NioEventLoop
NioEventLoop 表示一個不斷循環的執行處理任務的線程，每個NioEventLoop 都有一個selector , 用于監聽綁定在其上的socket的網絡通訊
NioEventLoopGroup 可以有多個線程, 即可以含有多個NioEventLoop
每個Boss NioEventLoop 循環執行的步驟有3步

輪詢accept 事件
處理accept 事件 , 與client建立連接配接 , 生成NioScocketChannel , 并将其注冊到某個worker NIOEventLoop 上的 selector
處理任務隊列的任務，即 runAllTasks

每個 Worker NIOEventLoop 循環執行的步驟

輪詢read, write 事件
處理i/o事件，即read , write 事件，在對應NioScocketChannel 處理
處理任務隊列的任務，即 runAllTasks

每個Worker NIOEventLoop 處理業務時，會使用pipeline(管道), pipeline 中包含了 channel , 即通過pipeline 可以擷取到對應通道, 管道中維護了很多的處理器

Netty快速入門執行個體-TCP服務

NettyServer

package com.xiaoliuliu.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * @author 小六六
 * @version 1.0
 * @date 2020/8/16 17:41
 */
public class NettyServer {

    public static void main(String[] args) {

            //建立BossGroup 和 WorkerGroup
            //說明
            //1. 建立兩個線程組 bossGroup 和 workerGroup
            //2. bossGroup 隻是處理連接配接請求 , 真正的和用戶端業務處理，會交給 workerGroup完成
            //3. 兩個都是無限循環
            //4. bossGroup 和 workerGroup 含有的子線程(NioEventLoop)的個數
            //   預設實際 cpu核數 * 2


            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);

            NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);

        try {
            //建立伺服器端的啟動對象，配置參數
            ServerBootstrap bootstrap = new ServerBootstrap();

            bootstrap.group(bossGroup, workerGroup)//設定兩個線程組
                    .channel(NioServerSocketChannel.class) //使用NioSocketChannel 作為伺服器的通道實作
                    .option(ChannelOption.SO_BACKLOG,128) //設定保持活動連接配接狀态
            //        .handler(null) // 該 handler對應 bossGroup , childHandler 對應 workerGroup
                    .childHandler(new ChannelInitializer<SocketChannel>() {//建立一個通道初始化對象(匿名對象)
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            System.out.println("客戶socketchannel hashcode=" + ch.hashCode()); //可以使用一個集合管理 SocketChannel， 再推送消息時，可以将業務加入到各個channel 對應的 NIOEventLoop 的 taskQueue 或者 scheduleTaskQueue
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    });


            System.out.println(".....伺服器 is ready...");


            //綁定一個端口并且同步, 生成了一個 ChannelFuture 對象
            //啟動伺服器(并綁定端口)
            //基于異步操作
            ChannelFuture cf = bootstrap.bind(6666).sync();


            //給cf 注冊監聽器，監控我們關心的事件
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (cf.isSuccess()) {
                        System.out.println("監聽端口 6666 成功");
                    } else {
                        System.out.println("監聽端口 6666 失敗");
                    }
                }
            });

        //對關閉通道進行監聽
            cf.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {

            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}

NettyServerHandler

package com.xiaoliuliu.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.util.CharsetUtil;

/**
 * @author 小六六
 * @version 1.0
 * @date 2020/8/16 18:04
 *
 *
 * 說明
 * 1. 我們自定義一個Handler 需要繼續netty 規定好的某個HandlerAdapter(規範)
 * 2. 這時我們自定義一個Handler , 才能稱為一個handler
 *
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {


    //讀取資料實際(這裡我們可以讀取用戶端發送的消息)
    /*
    1. ChannelHandlerContext ctx:上下文對象, 含有 管道pipeline , 通道channel, 位址
    2. Object msg: 就是用戶端發送的資料 預設Object
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        /*

        //比如這裡我們有一個非常耗時長的業務-> 異步執行 -> 送出該channel 對應的
        //NIOEventLoop 的 taskQueue中,

        //解決方案1 使用者程式自定義的普通任務

        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {

                try {
                    Thread.sleep(5 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 用戶端~(>^ω^<)喵2", CharsetUtil.UTF_8));
                    System.out.println("channel code=" + ctx.channel().hashCode());
                } catch (Exception ex) {
                    System.out.println("發生異常" + ex.getMessage());
                }
            }
        });

        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {

                try {
                    Thread.sleep(5 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 用戶端~(>^ω^<)喵3", CharsetUtil.UTF_8));
                    System.out.println("channel code=" + ctx.channel().hashCode());
                } catch (Exception ex) {
                    System.out.println("發生異常" + ex.getMessage());
                }
            }
        });

        //解決方案2 : 使用者自定義定時任務 -》 該任務是送出到 scheduleTaskQueue中

        ctx.channel().eventLoop().schedule(new Runnable() {
            @Override
            public void run() {

                try {
                    Thread.sleep(5 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 用戶端~(>^ω^<)喵4", CharsetUtil.UTF_8));
                    System.out.println("channel code=" + ctx.channel().hashCode());
                } catch (Exception ex) {
                    System.out.println("發生異常" + ex.getMessage());
                }
            }
        }, 5, TimeUnit.SECONDS);



        System.out.println("go on ...");*/
        System.out.println("伺服器讀取線程 " + Thread.currentThread().getName() + " channle =" + ctx.channel());
        System.out.println("server ctx =" + ctx);
        System.out.println("看看channel 和 pipeline的關系");

        Channel channel = ctx.channel();
        ChannelPipeline pipeline = channel.pipeline(); //本質是一個雙向連結, 出站入站

        ByteBuf byteBuf=(ByteBuf)msg;

        System.out.println("用戶端發送的資料是"+byteBuf.toString(CharsetUtil.UTF_8));

        System.out.println("用戶端位址:" + channel.remoteAddress());
    }


    //資料讀取完畢
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //writeAndFlush 是 write + flush
        //将資料寫入到緩存，并重新整理
        //一般講，我們對這個發送的資料進行編碼
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 用戶端~(>^ω^<)喵1", CharsetUtil.UTF_8));
    }

    //處理異常，關閉通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

NettyClient

package com.xiaoliuliu.netty.simple;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * @author 小六六
 * @version 1.0
 * @date 2020/8/16 18:13
 */
public class NettyClient {

    public static void main(String[] args) {
        //用戶端需要一個事件循環組
        EventLoopGroup group = new NioEventLoopGroup();

        try {

            //建立用戶端啟動對象
            //注意用戶端使用的不是 ServerBootstrap 而是 Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)//設定線程組
                    .channel(NioSocketChannel.class)// 設定用戶端通道的實作類(反射)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                           ch.pipeline().addLast(new NettyClientHandler());
                        }
                    });

            System.out.println("用戶端 ok..");

            //啟動用戶端去連接配接伺服器端
            //關于 ChannelFuture 要分析，涉及到netty的異步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();

            //給關閉通道進行監聽
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }


    }
}

NettyClientHandler

package com.xiaoliuliu.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * @author 小六六
 * @version 1.0
 * @date 2020/8/16 18:23
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    //當通道就緒就會觸發該方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client " + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 小六六: (>^ω^<)喵", CharsetUtil.UTF_8));
    }

    //當通道有讀取事件時，會觸發
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("伺服器回複的消息:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("伺服器的位址： "+ ctx.channel().remoteAddress());
    }


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

結果

結尾

Netty 抽象出兩組線程池，BossGroup 專門負責接收用戶端連接配接，WorkerGroup 專門負責網絡讀寫操作。
NioEventLoop 表示一個不斷循環執行處理任務的線程，每個 NioEventLoop 都有一個 selector，用于監聽綁定在其上的 socket 網絡通道。
NioEventLoop 内部采用串行化設計，從消息的讀取->解碼->處理->編碼->發送，始終由 IO 線程 NioEventLoop 負責
NioEventLoopGroup 下包含多個 NioEventLoop
每個 NioEventLoop 中包含有一個 Selector，一個 taskQueue
每個 NioEventLoop 的 Selector 上可以注冊監聽多個 NioChannel
每個 NioChannel 隻會綁定在唯一的 NioEventLoop 上
每個 NioChannel 都綁定有一個自己的 ChannelPipeline

小六六學Netty系列之初識Netty

前言

絮叨

Netty官網

Netty官網說明

Netty的優點

Reactor 模式

Netty模型

Netty快速入門執行個體-TCP服務

結尾

繼續閱讀

知名測網速工具，已解鎖全功能

對自己發展的一些思考

Debug時檢測到Loaderlock的解決辦法

淺談MFC圖示

.NET程式員的技能分析參考V1.0(有附件)

asp.net AspNetpager的用法

在angularjs中通過ts編寫時，如何添加中英文轉換

Java網絡程式設計-Socket程式設計初涉三（僞異步I/O模型的簡易多人聊天室）

msdn 訪談錄之C＃程式設計二

使用Windbg調試.Net應用程式

部署.net平台的應用程式

Robot Framework配置發送郵件功能

II6 IIS7 下 asp.net的生命周期 ASP.NET Application Life Cycle Overview for IIS 5.0 and 6.0

2010年終總結--技術篇

如何成為一名.net 工程師?

Netty——自定義協定解決TCP粘包拆包問題什麼是TCP粘包拆包自定義協定解決拆包粘包問題