Netty入門

時間 2019-11-16

標籤 netty 入門欄目 Netty 简体版

原文原文鏈接

一、什麼是Nettyjava

Netty 是一個基於 JAVA NIO 類庫的異步通訊框架，它的架構特色是：異步非阻塞、基於事件驅動、高性能、高可靠性和高可定製性。算法

二、Netty應用場景編程

1.分佈式開源框架中dubbo、Zookeeper，RocketMQ底層rpc通信使用就是netty。json

2.遊戲開發中，底層使用netty通信。bootstrap

三、爲何選擇netty數組

NIO的類庫和API繁雜，使用麻煩，你須要熟練掌握Selector、ServerSocketChannel、SocketChannel、ByteBuffer等；
須要具有其它的額外技能作鋪墊，例如熟悉Java多線程編程，由於NIO編程涉及到Reactor模式，你必須對多線程和網路編程很是熟悉，才能編寫出高質量的NIO程序；
JDK NIO的BUG，例如臭名昭著的epoll bug，它會致使Selector空輪詢，最終致使CPU 100%

四、netty案例安全

服務端服務器

package com.zhang.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class HelloServer {

    /**
     * 服務端監聽的端口地址
     */
    private static final int portNumber = 7878;

    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup);
            b.channel(NioServerSocketChannel.class);
            b.childHandler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    // 以("\n")爲結尾分割的 解碼器
                    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));

                    // 字符串解碼 和 編碼
                    pipeline.addLast("decoder", new StringDecoder());
                    pipeline.addLast("encoder", new StringEncoder());

                    // 本身的邏輯Handler
                    pipeline.addLast("handler", new HelloServerHandler());
                }
            });

            // 服務器綁定端口監聽
            ChannelFuture f = b.bind(portNumber).sync();
            // 監聽服務器關閉監聽
            f.channel().closeFuture().sync();

            // 能夠簡寫爲
            /* b.bind(portNumber).sync().channel().closeFuture().sync(); */
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

package com.zhang.netty;

import java.net.InetAddress;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class HelloServerHandler extends SimpleChannelInboundHandler<String> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        // 收到消息直接打印輸出
        System.out.println(ctx.channel().remoteAddress() + " Say : " + msg);
        
        // 返回客戶端消息 - 我已經接收到了你的消息
        ctx.writeAndFlush("Received your message："+msg+"\n");
    }
    
    /*
     * 覆蓋 channelActive 方法 在channel被啓用的時候觸發 (在創建鏈接的時候)
     * */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
        System.out.println("RamoteAddress : " + ctx.channel().remoteAddress() + " active !");
        
        ctx.writeAndFlush( "Welcome to " + InetAddress.getLocalHost().getHostName() + " service!\n");
        
        super.channelActive(ctx);
    }
}

客戶端網絡

package com.zhang.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class HelloClient {
    
    public static String host = "127.0.0.1";
    public static int port = 7878;

    /**
     * @param args
     * @throws InterruptedException 
     * @throws IOException 
     */
    public static void main(String[] args) throws InterruptedException, IOException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    //這個地方的 必須和服務端對應上。不然沒法正常解碼和編碼
                    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                    pipeline.addLast("decoder", new StringDecoder());
                    pipeline.addLast("encoder", new StringEncoder());

                    // 客戶端的邏輯
                    pipeline.addLast("handler", new HelloClientHandler());
                }
            });

            // 鏈接服務端
            Channel ch = b.connect(host, port).sync().channel();
            
            // 控制檯輸入
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            for (;;) {
                String line = in.readLine();
                if (line == null) {
                    continue;
                }
                /*
                 * 向服務端發送在控制檯輸入的文本 並用"\r\n"結尾
                 * 之因此用\r\n結尾 是由於咱們在handler中添加了 DelimiterBasedFrameDecoder 幀解碼。
                 * 這個解碼器是一個根據\n符號位分隔符的解碼器。因此每條消息的最後必須加上\n不然沒法識別和解碼
                 * */
                ch.writeAndFlush(line + "\r\n");
            }
        } finally {
            group.shutdownGracefully();
        }
    }
}

package com.zhang.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class HelloClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        
        System.out.println("Server say : " + msg);
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Client active ");
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Client close ");
        super.channelInactive(ctx);
    }
}

五、什麼是粘包/拆包多線程

一個完整的業務可能會被TCP拆分紅多個包進行發送，也有可能把多個小的包封裝成一個大的數據包發送，這個就是TCP的拆包和封包問題。下面能夠看一張圖，是客戶端向服務端發送包：

第一種狀況，Data1和Data2都分開發送到了Server端，沒有產生粘包和拆包的狀況。

第二種狀況，Data1和Data2數據粘在了一塊兒，打成了一個大的包發送到Server端，這個狀況就是粘包。

第三種狀況，Data2被分離成Data2_1和Data2_2，而且Data2_1在Data1以前到達了服務端，這種狀況就產生了拆包。

因爲網絡的複雜性，可能數據會被分離成N多個複雜的拆包/粘包的狀況，因此在作TCP服務器的時候就須要首先解決拆包

解決辦法：

消息定長，報文大小固定長度，不夠空格補全，發送和接收方遵循相同的約定，這樣即便粘包了經過接收方編程實現獲取定長報文也能區分。

sc.pipeline().addLast(new FixedLengthFrameDecoder(10));

包尾添加特殊分隔符，例如每條報文結束都添加回車換行符（例如FTP協議）或者指定特殊字符做爲報文分隔符，接收方經過特殊分隔符切分報文區分。

//使用特殊字符
ByteBuf buf = Unpooled.copiedBuffer("ha".getBytes());
sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));

//使用\n
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));

六、序列化定義

序列化（serialization）就是將對象序列化爲二進制形式（字節數組），通常也將序列化稱爲編碼（Encode），主要用於網絡傳輸、數據持久化等；

反序列化（deserialization）則是將從網絡、磁盤等讀取的字節數組還原成原始對象，以便後續業務的進行，通常也將反序列化稱爲解碼（Decode），主要用於網絡傳輸對象的解碼，以便完成遠程調用。

七、序列化協議

Java默認提供的序列化機制，須要序列化的Java對象只須要實現 Serializable / Externalizable 接口並生成序列化ID，這個類就可以經過 ObjectInput 和 ObjectOutput 序列化和反序列化。可是Java默認提供的序列化有不少問題，主要有如下幾個缺點：

沒法跨語言：由於Java序列化後的字節數組，其它語言沒法進行反序列化。；
序列化後的碼流太大:：相對於目前主流的序列化協議，Java序列化後的碼流太大；
序列化的性能差：因爲Java序列化採用同步阻塞IO，相對於目前主流的序列化協議，它的效率很是差。

八、影響序列化性能的關鍵因素

序列化後的碼流大小（網絡帶寬的佔用）；
序列化的性能（CPU資源佔用）；
是否支持跨語言（異構系統的對接和開發語言切換）

九、幾種流行的序列化協議比較

XML

XML（Extensible Markup Language）是一種經常使用的序列化和反序列化協議，它歷史悠久，從1998年的1.0版本被普遍使用至今。

優勢：人機可讀性好，可指定元素或特性的名稱

缺點：

序列化數據只包含數據自己以及類的結構，不包括類型標識和程序集信息。
類必須有一個將由 XmlSerializer 序列化的默認構造函數。
只能序列化公共屬性和字段
不能序列化方法
文件龐大，文件格式複雜，傳輸佔帶寬

使用場景：當作配置文件存儲數據，實時數據轉換

JSON

JSON(JavaScript Object Notation, JS 對象標記) 是一種輕量級的數據交換格式。它基於 ECMAScript (w3c制定的js規範)的一個子集， JSON採用與編程語言無關的文本格式，可是也使用了類C語言（包括C， C++， C#， Java， JavaScript， Perl， Python等）的習慣，簡潔和清晰的層次結構使得 JSON 成爲理想的數據交換語言。

優勢：