Netty粘包拆包及使用原理詳解

更新時間：2022年08月01日 11:00:52 作者：kaico2018

Netty是由JBOSS提供的一個java開源框架，現(xiàn)為?Github上的獨立項目。Netty提供異步的、事件驅(qū)動的網(wǎng)絡(luò)應(yīng)用程序框架和工具，用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò)服務(wù)器和客戶端程序，這篇文章主要介紹了Netty粘包拆包及使用原理

為什么使用Netty框架

NIO的類庫和API繁雜，使用麻煩，你需要熟練掌握Selector、ServerSocketChannel、SocketChannel、ByteBuffer等。
需要具備其他的額外技能做鋪墊，例如熟悉Java多線程編程。這是因為NIO編程涉及到 Reactor 模式，你必須對多線程和網(wǎng)路編程非常熟悉，才能編寫出高質(zhì)量的NIO程序。
可靠性能力補齊，工作量和難度都非常大。例如客戶端面臨斷連重連、網(wǎng)絡(luò)閃斷、半包讀寫、失敗緩存、網(wǎng)絡(luò)擁塞和異常碼流的處理等問題，NIO編程的特點是功能開發(fā)相對容易，但是可靠性能力補齊的工作量和難度都非常大。
JDK NIO的BUG，例如臭名昭著的 epoll bug，它會導(dǎo)致Selector空輪詢，最終導(dǎo)致CPU 100%。官方聲稱在JDK1.6版本的update18修復(fù)了該問題，但是直到JDK1.7版本該問題仍舊存在，只不過該BUG發(fā)生概率降低了一些而已，它并沒有被根本解決。該BUG以及與該BUG相關(guān)的問題單可以參見以下鏈接內(nèi)容。

由于上述原因，在大多數(shù)場景下，不建議大家直接使用JDK的NIO類庫，除非你精通NIO編程或者有特殊的需求。在絕大多數(shù)的業(yè)務(wù)場景中，我們可以使用NIO框架Netty來進行NIO編程，它既可以作為客戶端也可以作為服務(wù)端，同時支持UDP和異步文件傳輸，功能非常強大。

Netty框架介紹

Netty是業(yè)界最流行的NIO框架之一，它的健壯性、功能、性能、可定制性和可擴展性在同類框架中都是首屈一指的，它已經(jīng)得到成百上千的商用項目驗證，例如Hadoop的RPC框架Avro就使用了Netty作為底層通信框架，其他還有業(yè)界主流的RPC框架，也使用Netty來構(gòu)建高性能的異步通信能力。

優(yōu)點總結(jié):

API使用簡單，開發(fā)門檻低；
功能強大，預(yù)置了多種編解碼功能，支持多種主流協(xié)議；
定制能力強，可以通過ChannelHandler對通信框架進行靈活地擴展；
性能高，通過與其他業(yè)界主流的NIO框架對比，Netty的綜合性能最優(yōu)；
成熟、穩(wěn)定，Netty修復(fù)了已經(jīng)發(fā)現(xiàn)的所有JDK NIO BUG，業(yè)務(wù)開發(fā)人員不需要再為NIO的BUG而煩惱；
社區(qū)活躍，版本迭代周期短，發(fā)現(xiàn)的BUG可以被及時修復(fù)，同時，更多的新功能會加入；
經(jīng)歷了大規(guī)模的商業(yè)應(yīng)用考驗，質(zhì)量得到驗證。Netty在互聯(lián)網(wǎng)、大數(shù)據(jù)、網(wǎng)絡(luò)游戲、企業(yè)應(yīng)用、電信軟件等眾多行業(yè)已經(jīng)得到了成功商用，證明它已經(jīng)完全能夠滿足不同行業(yè)的商業(yè)應(yīng)用了。

Netty實戰(zhàn)

首先引入Netty的jar包。

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

Netty編寫服務(wù)器端

NettyServer 類

public class NettyServer {
    /**
     * netty啟動端口號
     */
    private static int port = 8080;
    public static void main(String[] args) {
        /**
         *  客戶端創(chuàng)建兩個線程池組分別為 boss線程組和工作線程組
         */
        // 用于接受客戶端連接的請求 （并沒有處理請求）
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        // 用于處理客戶端連接的讀寫操作(處理請求操作)
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        //NioServerSocketChannel   標(biāo)記當(dāng)前是服務(wù)器端
        serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 設(shè)置我們分割最大長度為1024
                        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        // 獲取數(shù)據(jù)的結(jié)果為string類型
                        socketChannel.pipeline().addLast(new StringEncoder());
                         //處理每個handler（也就是每次客戶端請求）
                        socketChannel.pipeline().addLast(new ServerHandler());
                    }
                });
        try {
            //綁定端口號
            ChannelFuture bind = serverBootstrap.bind(port);
            ChannelFuture sync = bind.sync();
            System.out.println("服務(wù)器端啟動成功：" + port);
            //等待監(jiān)聽我們的請求
            sync.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //優(yōu)雅的關(guān)閉我們的線程池
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

ServerHandler 類

public class ServerHandler extends SimpleChannelInboundHandler {
    /*
     * @Author kaico
     * @Date 9:56 2020/10/8
     * @Description //TODO 獲取數(shù)據(jù)
     * @Param [channelHandlerContext, o]
     * @return void
     **/
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        ByteBuf byteBuf = (ByteBuf) o;
        String request = byteBuf.toString(CharsetUtil.UTF_8);
        System.out.println("request:" + request);
        // 響應(yīng)內(nèi)容:
        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("這里是Netty服務(wù)端\n", CharsetUtil.UTF_8));
    }
}

Netty客戶端

NettyClient 類

public class NettyClient {
    public static void main(String[] args) {
        //創(chuàng)建nioEventLoopGroup
        NioEventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group).channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress("127.0.0.1", 8080))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 設(shè)置我們分割最大長度為1024
                        ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        // 獲取數(shù)據(jù)的結(jié)果為string類型
                        ch.pipeline().addLast(new StringEncoder());
                        ch.pipeline().addLast(new ClientHandler());
                    }
                });
        try {
            // 發(fā)起同步連接
            ChannelFuture sync = bootstrap.connect().sync();
            sync.channel().closeFuture().sync();
        } catch (Exception e) {

        } finally {
            group.shutdownGracefully();
        }
    }
}

ClientHandler 類

public class ClientHandler extends SimpleChannelInboundHandler {
    /**
     * 活躍通道可以發(fā)送消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 10; i++) {
            // 發(fā)送數(shù)據(jù)
            ctx.writeAndFlush(Unpooled.copiedBuffer("你是什么類型的服務(wù)端啊?\n", CharsetUtil.UTF_8));
        }
        //客戶端發(fā)十條消息
    }
    /**
     * 讀取消息
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("resp：" + byteBuf.toString(CharsetUtil.UTF_8));
    }
}

粘包與拆包

原因：因為我們現(xiàn)在tcp協(xié)議默認(rèn)是長連接形式實現(xiàn)通訊，發(fā)送請求完了之后整個連接暫時不會關(guān)閉

1.短連接

客戶端與服務(wù)器端建立連接的時候，客戶端發(fā)送一條消息，客戶端與服務(wù)器連接關(guān)閉

2.長連接

客戶端與服務(wù)器端建立連接的時候，客戶端發(fā)送多條消息，客戶端與服務(wù)器連接關(guān)閉

什么是粘包：多次發(fā)送的消息，服務(wù)器一次合并讀取msgmsg

什么是拆包：多次發(fā)送消息服務(wù)器讀取第一條數(shù)據(jù)完整+第二條不完整數(shù)據(jù) 第二條不完整數(shù)據(jù) Msgm sg

為什么會造成拆包和粘包? 前提長連接、其次緩沖區(qū)

原因的造成：

Tcp協(xié)議為了能夠高性能的傳輸數(shù)據(jù)，發(fā)送和接受時候都會采用緩沖區(qū)，必須等待緩沖區(qū)滿了以后才可以發(fā)送或者讀??；

當(dāng)我們的應(yīng)用程序如果發(fā)送的數(shù)據(jù)大于了我們的套字節(jié)的緩沖區(qū)大小的話，就會造成了拆包。拆分成多條消息讀取。當(dāng)我們應(yīng)用程序如果發(fā)送的寫入的消息如果小于套字節(jié)緩沖區(qū)大小的時候

粘包與拆包產(chǎn)生的背景：

Tcp協(xié)議為了高性能的傳輸，發(fā)送和接受的時候都采用了緩沖區(qū)

3. 當(dāng)我們的應(yīng)用程序發(fā)送的數(shù)據(jù)大于套字節(jié)緩沖區(qū)的時候，就會實現(xiàn)拆包。

4. 當(dāng)我們的應(yīng)用程序?qū)懭氲臄?shù)據(jù)小于套字節(jié)緩沖區(qū)的時候，多次發(fā)送的消息會合并到一起接受，這個過程我們可以稱做為粘包。

5. 接受端不夠及時的獲取緩沖區(qū)的數(shù)據(jù)，也會產(chǎn)生粘包的問題

6. 進行mss（最大報文長度）大小的TCP分段，當(dāng)TCP報文長度-TCP頭部長度>mss的時候?qū)l(fā)生拆包。

解決思路：

7. 以固定的長度發(fā)送數(shù)據(jù)，到緩沖區(qū)

8. 可以在數(shù)據(jù)之間設(shè)置一些邊界(\n或者\r\n)

9. 利用編碼器LineBaseDFrameDecoder解決tcp粘包的問題

常用編碼器：

DelimiterBasedFrameDecoder 解決TCP的粘包解碼器
StringDecoder 消息轉(zhuǎn)成String解碼器
LineBasedFrameDecoder 自動完成標(biāo)識符分隔解碼器
FixedLengthFrameDecoder 固定長度解碼器，二進制
Base64Decoder 解碼器

利用編碼器LineBaseDFrameDecoder解決tcp粘包的問題的Java代碼案例，核心思路就是增加邊界 \n

服務(wù)器端類 NettyServer 的修改點

serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                // 設(shè)置我們分割最大長度為1024
                socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                // 獲取數(shù)據(jù)的結(jié)果為string類型
                socketChannel.pipeline().addLast(new StringEncoder());
                //發(fā)送數(shù)據(jù)的時候設(shè)置邊界 \n
                socketChannel.pipeline().addLast(new ServerHandler());
            }
        });

服務(wù)器端類 ServerHandler 的修改點

@Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        ByteBuf byteBuf = (ByteBuf) o;
        String request = byteBuf.toString(CharsetUtil.UTF_8);
        System.out.println("request:" + request);
        // 響應(yīng)內(nèi)容:
        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("這里是Netty服務(wù)端\n", CharsetUtil.UTF_8));
    }

客戶端的類 NettyClient 的修改點

bootstrap.group(group).channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress("127.0.0.1", 8080))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 設(shè)置我們分割最大長度為1024
                        ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        // 獲取數(shù)據(jù)的結(jié)果為string類型
                        ch.pipeline().addLast(new StringEncoder());
                        ch.pipeline().addLast(new ClientHandler());
                    }
                });

客戶端的類 ClientHandler 的修改點

 @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 10; i++) {
            // 發(fā)送數(shù)據(jù)
            ctx.writeAndFlush(Unpooled.copiedBuffer("你是什么類型的服務(wù)端啊?\n", CharsetUtil.UTF_8));
        }
        //客戶端發(fā)十條消息
    }

到此這篇關(guān)于Netty粘包拆包詳解及實戰(zhàn)流程的文章就介紹到這了,更多相關(guān)Netty粘包拆包內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: