閉關修煉（十三）Netty5.0

文章目錄

• Netty5
• • jar包
  • 服務端
  • 用戶端
  • 長連接配接與短連接配接
• 粘包與拆包
• • 什麼是粘包和拆包
  • 拆包方法
• Netty 5.0架構剖析和源碼解讀.pdf

Netty5

Netty有很多版本，在4的時候就有巨大的革命，API基本都改了。

官方說5不安全。開發中用Netty4比較多

TCP粘包、拆包用netty3代碼特别複雜，netty4隻用分隔符進行區分就行了，比較容易一些。

jar包

<dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>5.0.0.Alpha2</version>
        </dependency>
        <dependency>
            <groupId>org.jboss.marshalling</groupId>
            <artifactId>jboss-marshalling</artifactId>
            <version>1.3.19.GA</version>
        </dependency>
        <dependency>
            <groupId>org.jboss.marshalling</groupId>
            <artifactId>jboss-marshalling-serial</artifactId>
            <version>1.3.18.GA</version>
            <scope>test</scope>
        </dependency>
           

服務端

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;

class ServerHandler extends ChannelHandlerAdapter {
    // 當通道被調用時會執行的方法（拿到資料）
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
        // 用戶端的msg
        String value = (String)msg;
        System.out.println("伺服器收到用戶端消息：" + value);
        // 回複用戶端, 加入到緩沖區并發送
        ctx.writeAndFlush("伺服器收到");
        // 上面writeAndFlush等同于先調用write再調用flush
    }
}
public class NettyServer {
    static int PORT = 8080;
    public static void main(String[] args) throws InterruptedException {
        System.out.println("服務端啟動");
        // 建立兩個線程池，一個負責接收用戶端，一個進行傳輸
        NioEventLoopGroup pGroup = new NioEventLoopGroup();
        NioEventLoopGroup cGroup = new NioEventLoopGroup();
        // 建立輔助類
        ServerBootstrap sbs = new ServerBootstrap();
        sbs.group(pGroup, cGroup).channel(NioServerSocketChannel.class)
                // 設定緩沖區大小
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_SNDBUF, 32 * 1024)
                .option(ChannelOption.SO_RCVBUF, 32 * 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 設定放回結果是string類型
                        socketChannel.pipeline().addLast(new StringDecoder());
                        socketChannel.pipeline().addLast(new ServerHandler());
                    }
                });
        ChannelFuture cf = sbs.bind(PORT).sync();
        // 關閉
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
        cGroup.shutdownGracefully();
    }
}

           

用戶端

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;

class ClientHandler extends ChannelHandlerAdapter{
    // 當通道被調用執行該方法
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
        // 接收資料
        String data = (String) msg;
        System.out.println("client msg:" + data);
    }
}
public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("用戶端啟動");
        NioEventLoopGroup pGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(pGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringDecoder());
                        socketChannel.pipeline().addLast(new ClientHandler());

                    }
                });
        ChannelFuture cf = bootstrap.connect("127.0.0.1", NettyServer.PORT).sync();
        // 發送兩次請求，自動設定了粘包
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("測試消息~".getBytes()));
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("測試消息~".getBytes()));
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("測試消息~".getBytes()));
        Thread.sleep(1000);
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("測試消息~".getBytes()));
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("測試消息~".getBytes()));

        // 等待用戶端端口号關閉
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
    }
}

           

長連接配接與短連接配接

關閉用戶端，伺服器會報IOException 遠端主機強迫關閉了一個現有的連接配接，是個是因為Netty5連接配接是長連接配接，沒有4次揮手直接關閉了用戶端。

長連接配接常用在移動端消息推送和MQ，短連接配接用于HTTP協定

粘包與拆包

什麼是粘包和拆包

TCP中有這個概念，粘包是将多個包合在一起，拆包是将一個包拆分多個包

在Netty5中已經幫我們做了優化傳輸，減輕伺服器傳輸壓力。

拆包方法

消息定長，封包大小固定長度，不夠空格補全，發送和接收放遵守相同的約定，這樣即使粘包了通過接收方程式設計實作擷取定長封包也能區分

在用戶端和服務端initChannel中添加：

另一種解決辦法是包尾添加特殊分隔符，例如每條封包結束都添加回車換行符或者指定特殊字元作為封包分隔符，接收方通過分隔符切分封包區分，如果用戶端writeAndFlush發送資料不包含分隔符，通道會被阻塞，不會發送包，直到writeAndFlush寫入通道的内容有分隔符，之前的資料一起被發送。

在用戶端和服務端initChannel中添加下面代碼，表示用_&進行分割

ByteBuf buf = Unpooled.copiedBuffer("_&".getBytes());
socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
           

還有一種是定義消息頭和消息體，消息頭包含資訊的總長度，不會，略。

Netty 5.0架構剖析和源碼解讀.pdf

https://pan.baidu.com/s/1mg4rN7M