netty

cfanr 發(fā)布于2019-08-15 13:34 / 3016人閱讀

摘要：設置每個數據包的大小如個字節(jié)，如果某個數據包不足個字節(jié)可能會出現丟包的情況，即該數據包未從一個端到另一個端，此時需要用空格或者既定的符號補充在數據包之間使用一些字符進行分割如號之類的，解析的時候先處理掉分隔符再拿到各個數據包就好了。

netty

概念: Netty是由JBOSS提供的一個java開源框架。Netty提供異步的、事件驅動的網絡應用程序框架和工具，用以快速開發(fā)高性能、高可靠性的網絡服務器和客戶端程序。
也就是說，Netty 是一個基于NIO的客戶、服務器端編程框架，使用Netty 可以確保你快速和簡單的開發(fā)出一個網絡應用，例如實現了某種協議的客戶，服務端應用。Netty相當簡化和流線化了網絡應用的編程開發(fā)過程，例如，TCP和UDP的socket服務開發(fā)。

新特性

處理大容量數據流更簡單

處理協議編碼和單元測試更簡單

I/O超時和idle狀態(tài)檢測

應用程序的關閉更簡單，更安全

更可靠的OutOfMemoryError預防

性能

更好的吞吐量，更低的延遲

更少的資源消耗

最小化不必要的內存拷貝

具體使用見代碼及注釋

Helloword版
服務端這邊綁定了兩個端口，可以根據業(yè)務區(qū)別對待如端口1是做A業(yè)務，端2做B業(yè)務.

public class Server {
    public static void main(String[] args) throws InterruptedException {
        //1.創(chuàng)建兩個線程組 (只有服務器端需要 )
        //一個線程組專門用來管理接收客戶端的請求連接的
        //一個線程組進行網絡通信(讀寫)
        EventLoopGroup receiveGroup = new NioEventLoopGroup();
        EventLoopGroup dealGroup = new NioEventLoopGroup();
        //創(chuàng)建輔助工具類，用于設置服務器通道的一系列配置
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(receiveGroup, dealGroup)//綁定兩個線程組
        .channel(NioServerSocketChannel.class)   //指定NIO的模式
        .option(ChannelOption.SO_BACKLOG, 1024)     //設置tcp緩沖區(qū)
        .option(ChannelOption.SO_SNDBUF, 32*1024) //設置發(fā)送緩沖區(qū)大小
        .option(ChannelOption.SO_RCVBUF, 32*1024) //設置接收緩沖大小
        .option(ChannelOption.SO_KEEPALIVE, true)  //保持連接
        .childHandler(new ChannelInitializer() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                //3 在這里配置具體數據接收方法的處理
                sc.pipeline().addLast(new ServerHandler());
            }
        });
        //4 進行綁定 
                ChannelFuture cf1 = serverBootstrap.bind(8765).sync();
                ChannelFuture cf2 = serverBootstrap.bind(8764).sync();
                //5 等待關閉
                cf1.channel().closeFuture().sync();
                cf2.channel().closeFuture().sync();
                receiveGroup.shutdownGracefully();
                dealGroup.shutdownGracefully();
     }
}

服務端處理器:

public class ServerHandler extends ChannelHandlerAdapter {


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("server channel active... ");
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "gbk");
            System.out.println("Server :" + body );
            String response = "進行返回給客戶端的響應：" + body ;
            //注意使用了writeAndFlush的話就可以不釋放ReferenceCountUtil.release(msg); 否則需要釋放ByteBuf容器的數據。
            ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
            //.addListener(ChannelFutureListener.CLOSE);//監(jiān)聽，內容傳輸完畢后就關閉管道
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)
            throws Exception {
        System.out.println("讀完了");
        ctx.flush();
    }

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


}

客戶端:

public class Client {
public static void main(String[] args) throws Exception{
        
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                sc.pipeline().addLast(new ClientHandler());
            }
        });
        
        ChannelFuture cf1 = b.connect("127.0.0.1", 8765).sync();
        ChannelFuture cf2 = b.connect("127.0.0.1", 8764).sync();
        //發(fā)送消息
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("C1：777".getBytes()));
        Thread.sleep(1000);
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("C1:666".getBytes()));
        cf2.channel().writeAndFlush(Unpooled.copiedBuffer("C2:888".getBytes()));
        Thread.sleep(2000);
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("C1:888".getBytes()));
        cf2.channel().writeAndFlush(Unpooled.copiedBuffer("C2:666".getBytes()));
        
        cf1.channel().closeFuture().sync();
        cf2.channel().closeFuture().sync();
        group.shutdownGracefully();
        
        
        
    }
}

客戶端處理器：

public class ClientHandler extends ChannelHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客戶端的channelActive()方法");
    }

    @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, "gbk");
            System.out.println("Client :" + body );
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

TCP拆包粘包問題

TCP是個“流”協議，所謂流，就是沒有界限的一串數據。大家可以想想河里的流水，是連成一片的，其間并沒有分界線。TCP底層并不了解上層業(yè)務數據的具體含義，它會根據TCP緩沖區(qū)的實際情況進行包的劃分，所以在業(yè)務上認為，一個完整的包可能會被TCP拆分成多個包進行發(fā)送，也有可能把多個小的包封裝成一個大的數據包發(fā)送，這就是所謂的TCP粘包和拆包問題。
通俗意義來說可能是三個數據如"A","B","C" 但經過TCP協議流式傳輸后成了"AB","C"兩個數據了,這種就是粘包了數據包之間粘一起了。那么拆包的話有三種方式。

設置每個數據包的大小如200個字節(jié)，如果某個數據包不足200個字節(jié)可能會出現丟包的情況，即該數據包未從一個端到另一個端，此時需要用空格或者既定的符號補充.

在數據包之間使用一些字符進行分割如$號之類的，解析的時候先處理掉分隔符再拿到各個數據包就好了。(一般用的比較多)

細粒化數據包分為頭和尾(將消息分為消息頭和消息尾)

其他

兩根水管（服務器與客戶端）需要相互流通水(數據)，那么需要一個轉接頭(套接字)連接，水流式無法區(qū)分一段段的數據，一種方式在流通的過程中設置些標志性物品如記號筆勾一下(分隔符)，另一種方式則是設定每一段都是多少容量的水來區(qū)分.

使用分隔符解決TCP粘包

可以理解管道流里流的都是ByteBuffer類型的數據，那么使用分隔符(非ByteBuffer類型)的話可能就意味著一個轉碼與解碼的過程。
服務端:

public class Server {
    public static void main(String[] args) throws Exception{
        //1 創(chuàng)建2個線程，一個是負責接收客戶端的連接。一個是負責進行數據傳輸的
        EventLoopGroup pGroup = new NioEventLoopGroup();
        EventLoopGroup cGroup = new NioEventLoopGroup();
        
        //2 創(chuàng)建服務器輔助類
        ServerBootstrap b = new ServerBootstrap();
        b.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() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                //設置特殊分隔符  解決TCP拆包黏包問題，
                ByteBuf buf = Unpooled.copiedBuffer("$".getBytes());
                sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                //設置字符串形式的解碼
                sc.pipeline().addLast(new StringDecoder());
                sc.pipeline().addLast(new ServerHandler());
            }
        });
        
        //4 綁定連接
        ChannelFuture cf = b.bind(8765).sync();
        
        //等待服務器監(jiān)聽端口關閉
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
        cGroup.shutdownGracefully();
        
    }
}

服務端處理器:

public class ServerHandler extends ChannelHandlerAdapter{
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(" server channel active... ");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String request = (String)msg;
        System.out.println("Server channelRead:" + request);
        String response = "服務器響應：" + msg + "$";
        ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelReadComplete");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
        System.out.println("exceptionCaught");
        ctx.close();
    }

}

客戶端：

public class Client {
public static void main(String[] args) throws Exception {
        
        EventLoopGroup group = new NioEventLoopGroup();
        
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(NioSocketChannel.class)
         .handler(new ChannelInitializer() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                //
                ByteBuf buf = Unpooled.copiedBuffer("$".getBytes());
                sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                sc.pipeline().addLast(new StringDecoder());
                sc.pipeline().addLast(new ClientHandler());
            }
        });
        
        ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
        
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("數據A$".getBytes()));
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("數據B$".getBytes()));
        
        
        //等待客戶端端口關閉
        cf.channel().closeFuture().sync();
        group.shutdownGracefully();
        
    }
}

客戶端處理器:

public class ClientHandler extends ChannelHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client channel active... ");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            String response = (String)msg;
            System.out.println("Client: " + response);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelReadComplete");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("exceptionCaught");
        ctx.close();
    }

}

設置長度大小解決TCP拆包黏包問題

服務端:

public class Server {

    public static void main(String[] args) throws Exception{
        //1 創(chuàng)建2個線程，一個是負責接收客戶端的連接。一個是負責進行數據傳輸的
        EventLoopGroup pGroup = new NioEventLoopGroup();
        EventLoopGroup cGroup = new NioEventLoopGroup();
        
        //2 創(chuàng)建服務器輔助類
        ServerBootstrap b = new ServerBootstrap();
        b.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() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                //設置定長字符串接收
                sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
                //設置字符串形式的解碼
                sc.pipeline().addLast(new StringDecoder());
                sc.pipeline().addLast(new ServerHandler());
            }
        });
        
        //4 綁定連接
        ChannelFuture cf = b.bind(8765).sync();
        
        //等待服務器監(jiān)聽端口關閉
        cf.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
        cGroup.shutdownGracefully();
        
    }
    
}

客戶端:

public class Client {

    public static void main(String[] args) throws Exception {
        
        EventLoopGroup group = new NioEventLoopGroup();
        
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(NioSocketChannel.class)
         .handler(new ChannelInitializer() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
                sc.pipeline().addLast(new StringDecoder());
                sc.pipeline().addLast(new ClientHandler());
            }
        });
        
        ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
        
        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("aaaaabbbbb".getBytes()));
        cf.channel().writeAndFlush(Unpooled.copiedBuffer("ccccccc".getBytes()));
        
        //等待客戶端端口關閉
        cf.channel().closeFuture().sync();
        group.shutdownGracefully();
        
    }
}

服務端與客戶端的處理器參照上例以字符串分割的.

                                新手上路，多多關注...

GPU云服務器云服務器 Netty netty 同步 webrtc netty netty webrtc

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規(guī)行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/68178.html

netty搭建web聊天室（1）

摘要：提供異步的事件驅動的網絡應用程序框架和工具，用以快速開發(fā)高性能高可靠性的網絡服務器和客戶端程序。總結我們完成了服務端的簡單搭建，模擬了聊天會話場景。之前一直在搞前端的東西，都快忘了自己是個java開發(fā)。其實還有好多java方面的東西沒搞過，突然了解到netty，覺得有必要學一學。介紹 Netty是由JBOSS提供的一個java開源框架。Netty提供異步的、事件驅動的網絡應用程序框...

izhuhaodev 2019-08-16 14:37 評論0 收藏0
慕課網_《Netty入門之WebSocket初體驗》學習總結

時間：2018年04月11日星期三說明：本文部分內容均來自慕課網。@慕課網：https://www.imooc.com 教學源碼：https://github.com/zccodere/s... 學習源碼：https://github.com/zccodere/s... 第一章：課程介紹 1-1 課程介紹什么是Netty 高性能、事件驅動、異步非阻塞的IO Java開源框架基于NIO的客戶...

Noodles 2019-08-15 15:30 評論0 收藏0
Netty 源碼分析之零磨刀不誤砍柴工源碼分析環(huán)境搭建

摘要：目錄此文章屬于源碼之下無秘密做最好的源碼分析教程系列文章之一代碼下載首先到的倉庫中點擊右邊綠色的按鈕拷貝地址然后在終端中輸入如下命令克隆工程工程源碼較大加上國內網絡問題下載源碼可能會比較耗時當有如下輸出時表示克隆成功了如果有朋友實在下載太目錄此文章屬于源碼之下無秘密 ── 做最好的 Netty 源碼分析教程系列文章之一. 代碼下載首先到 Netty 的 Github 倉庫中...

freewolf 2019-08-14 16:37 評論0 收藏0
源碼之下無秘密 ── 做最好的 Netty 源碼分析教程

摘要：背景在工作中雖然我經常使用到庫但是很多時候對的一些概念還是處于知其然不知其所以然的狀態(tài)因此就萌生了學習源碼的想法剛開始看源碼的時候自然是比較痛苦的主要原因有兩個第一網上沒有找到讓我滿意的詳盡的源碼分析的教程第二我也是第一次系統地學習這么大代背景在工作中, 雖然我經常使用到 Netty 庫, 但是很多時候對 Netty 的一些概念還是處于知其然, 不知其所以然的狀態(tài), 因此就萌生了學...

shenhualong 2019-08-14 16:37 評論0 收藏0
netty 基于 protobuf 協議實現 websocket 版本的簡易客服系統

摘要：結構作為服務端作為序列化數據的協議前端通訊演示地址服務端實現啟動類長連接示例主線程組從線程組請求的解碼和編碼把多個消息轉換為一個單一的或是，原因是解碼器會在每個消息中生成多個消息對象主要用于處理大數據流，比如一個大小的文件如果你直接傳輸肯定結構 netty 作為服務端 protobuf 作為序列化數據的協議 websocket 前端通訊演示 GitHub 地址 showImg(...

wua_wua2012 2019-08-23 14:11 評論0 收藏0