Netty高级应用及聊天室实战

Netty 高级应用

1. 编解码器

概念：在网络应用中，需要实现某种编解码器。将原始字节数据与自定义消息数据进行相互转换。网络中都是以字节码的形式传输的。

对Netty而言，编解码器由两部分组成：编码器、解码器

• 编码器：将消息对象转为字节或其他序列形式在网络上传输
• 解码器：负责将字节或其他序列形式转为指定的消息对象

Netty的编解码器实现了ChannelHandlerAdapter，也是一种特殊的ChannelHandler，所以依赖与ChannelPipeline，可以将多个编解码器链接在一起，以实现复杂的转换逻辑。

1. 解码器

• ByteToMessageDecoder:用于将字节转为消息，需要检查缓冲区是否有足够的字节
• ReplayingDecoder：继承ByteToMessageDecoder，不需要检查缓冲区是否有足够的字节,但是ReplayingDecoder速度略慢于ByteToMessageDecoder,同时不是所有的ByteBuf都支持。项目复杂性高则使用ReplayingDecoder，否则使用ByteToMessageDecode
• MessageToMessageDecoder：用于从一种消息解码为另一种消息（如POJO到POJO）

解码器示例：

public class DemoDecoder extends MessageToMessageDecoder<ByteBuf> {        @Override    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {        String msg = byteBuf.toString(CharsetUtil.UTF_8);        list.add(msg);    }}

通道里加入解码器：

 protected void initChannel(SocketChannel socketChannel) throws Exception {                        socketChannel.pipeline().addLast(new DemoDecoder());                        socketChannel.pipeline().addLast(new DemoNettyServerHandle());                    }

2. 编码器

• MessageToByteEncoder：将消息转为字节
• MessageToMessageEncoder：用于从一种消息编码为另外一种消息（例如POJO到POJO）

编码器示例：

public class DemoEncoder extends MessageToMessageEncoder<String> {    @Override    protected void encode(ChannelHandlerContext channelHandlerContext, String s, List<Object> list) throws Exception {        list.add(Unpooled.copiedBuffer(s,CharsetUtil.UTF_8));    }}

3. 编码解码器Codec

同时具备编码与解码功能

• ByteToMessageCodec
• MessageToMessageCodec

2. 基于Netty的HTTP服务器开发

效果如图：

代码如下：

public class NettyHttpServer {    private int port;    public NettyHttpServer(int port) {        this.port = port;    }    public static void main(String[] args) {        new NettyHttpServer(8090).run();    }    public void run(){        EventLoopGroup bossGroup=null;        EventLoopGroup workerGroup=null;        try{            bossGroup=new NioEventLoopGroup(1);            workerGroup=new NioEventLoopGroup();            ServerBootstrap serverBootstrap=new ServerBootstrap();            serverBootstrap.group(bossGroup,workerGroup)                    .channel(NioServerSocketChannel.class)                    .option(ChannelOption.SO_BACKLOG,128)                    .childOption(ChannelOption.SO_KEEPALIVE,Boolean.TRUE)                    .childHandler(new ChannelInitializer<SocketChannel>() {                        @Override                        protected void initChannel(SocketChannel socketChannel) throws Exception {                            //添加编解码器                            socketChannel.pipeline().addLast(new HttpServerCodec());                            socketChannel.pipeline().addLast(new NettyHttpServerHandler());                        }                    });            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();            channelFuture.channel().closeFuture().sync();        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();        }    }}

public class NettyHttpServerHandler extends SimpleChannelInboundHandler<HttpObject> {    @Override    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject) throws Exception {        if(httpObject instanceof HttpRequest){            DefaultHttpRequest request=(DefaultHttpRequest)httpObject;            if(request.uri().equals("/favicon.ico")){                //图标不响应                return;            }            System.out.println("接收到请求："+request.uri());            ByteBuf byteBuf = Unpooled.copiedBuffer("你好，我是服务端", CharsetUtil.UTF_8);            DefaultFullHttpResponse response=new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,HttpResponseStatus.OK,byteBuf);            //设置响应头            response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/html;charset=utf-8");            response.headers().set(HttpHeaderNames.CONTENT_LENGTH,byteBuf.readableBytes());            channelHandlerContext.writeAndFlush(response);        }    }}

3. 粘包和拆包

简介：粘包和拆包是TCP网络编程中不可避免的，无论客户端还是服务端，当我们读取或发送消息的时候都要考虑TCP底层的粘包/拆包机制。

粘包产生的原因：

• 应用程序写入的数据小于套接字缓冲区大小，网卡将应用多次写入的数据发送到网络上
• 接收方不及时读取套接字缓冲区数据
• TCP默认使用Nagle算法，将小数据包合并

拆包产生的原因：

• 数据太大超过剩余缓冲区的大小
• 数据太大超过MSS最大报文长度

粘包和拆包的解决方案

1. 消息长度固定，累计读取到定长的报文就认为是一个完整的信息
2. 将换行符作为消息结束符
3. 将特殊的分隔符作为消息的结束标志
4. 通过在消息头中定义长度字段来标识消息总长度

Netty中粘包和拆包的解决方案

Netty提供了4种解码器来解决：

1. 固定长度拆包器FixedLengthFrameDecoder
2. 行拆包器LineBasedFrameDecoder，以换行符作为分隔符
3. 分隔符拆包器DelimiterBasedFrameDecoder，通过自定义的分隔符进行拆分
4. 基于数据包长度的拆包器LengthFieldBasedFrameDecoder，将应用层数据包的长度最为拆分一句。要求应用层协议中包含数据包的长度。

DelimiterBasedFrameDecoder示例：

ByteBuf byteBuf =Unpooled.copiedBuffer("$".getBytes(StandardCharsets.UTF_8));ch.pipeline().addLast(new DelimiterBasedFrameDecoder(2048, byteBuf));

LengthFieldBasedFrameDecoder构造器参数讲解：
public LengthFieldBasedFrameDecoder(
ByteOrder byteOrder,
int lengthFieldOffset,
int lengthFieldLength,
int lengthAdjustment,
int initialBytesToStrip,
boolean failFast)

• byteOrder是指明Length字段是大端序还是小端序，因为Netty要读取Length字段的值，所以大端小端要设置好，默认Netty是大端序ByteOrder.BIG_ENDIAN。

• maxFrameLength是指最大包长度，如果Netty最终生成的数据包超过这个长度，Netty就会报错。

• lengthFieldOffset是指明Length的偏移位

• lengthFieldLength是Length字段长度

• lengthAdjustment 这个参数很多时候设为负数，这是最让小伙伴们迷惑的。下面我用一整段话来解释这个参数

当Netty利用lengthFieldOffset（偏移位）和lengthFieldLength（Length字段长度）成功读出Length字段的值后，Netty认为这个值是指从Length字段之后，到包结束一共还有多少字节，如果这个值是13，那么Netty就会再等待13个Byte的数据到达后，拼接成一个完整的包。但是更多时候，Length字段的长度，是指整个包的长度，如果是这种情况，当Netty读出Length字段的时候，它已经读取了包的4个Byte的数据，所以，后续未到达的数据只有9个Byte，即13 - 4 = 9，这个时候，就要用lengthAdjustment来告诉Netty，后续的数据并没有13个Byte，要减掉4个Byte，所以lengthAdjustment要设为 -4！！！

• initialBytesToStrip，跳过的个数。比如这里initialBytesToStrip设置为4，那么Netty就会跳过前4位解析后面的内容

• failFast 参数一般设置为true，当这个参数为true时，netty一旦读到Length字段，并判断Length超过maxFrameLength，就立即抛出异常。

示例：

 @Override    public void channelActive(ChannelHandlerContext channelHandlerContext) throws Exception {        for (int i=0;i<100;i++){            byte[] bytes = "你好，我是客户端".getBytes(CharsetUtil.UTF_8);            ByteBuf byteBuf = Unpooled.buffer();            byteBuf.writeInt(bytes.length);            byteBuf.writeBytes(bytes);            channelHandlerContext.writeAndFlush(byteBuf);        }    }

第2个参数和第三个参数表示：0-4个字节是内容长度字段，第五个参数的4代表跳过前4个字节。

 socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(60535,0,4,0,4));

最后输出的内容：

4. 基于Netty和WebSocket的聊天室案例

1. WebSocket简介

WebSocket是一种在单个TCP连接上进行全双工通信的协议。相比HTTP协议，WebSocket具备如下特点：

1. 支持双向通信，实时性更强
2. 更好的二进制支持
3. 较少的开销：协议控制的数据包头部较小

应用场景：

• 社交订阅
• 协同编辑
• 股票基金报价
• 体育实况更新
• 多媒体聊天
• 在线教育

2. 服务端开发

1. 引入依赖

基于SpringBoot环境

 <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency>        <!--添加thymeleaf依赖 -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-thymeleaf</artifactId>        </dependency>        <dependency>            <groupId>io.netty</groupId>            <artifactId>netty-all</artifactId>            <version>4.1.72.Final</version>        </dependency>

2. 核心后端代码

@Componentpublic class NettyWebSocketServer implements Runnable {    @Autowired    private NettyConfig nettyConfig;    @Autowired    private WebSocketChannelInit webSocketChannelInit;    private EventLoopGroup bossGroup = new NioEventLoopGroup(1);    private EventLoopGroup wokerGroup = new NioEventLoopGroup();    @PreDestroy    public void close(){        bossGroup.shutdownGracefully();        wokerGroup.shutdownGracefully();    }    @Override    public void run() {        try{            ServerBootstrap serverBootstrap=new ServerBootstrap();            serverBootstrap.group(bossGroup,wokerGroup)                    .channel(NioServerSocketChannel.class)                    .childHandler(webSocketChannelInit);            ChannelFuture channelFuture = serverBootstrap.bind(nettyConfig.getPort()).sync();            System.out.println("Netty服务端启动成功");            channelFuture.channel().closeFuture().sync();        }catch (Exception e){            e.printStackTrace();        }finally {            bossGroup.shutdownGracefully();            wokerGroup.shutdownGracefully();        }    }}

@Componentpublic class WebSocketChannelInit extends ChannelInitializer {    @Autowired    private NettyConfig nettyConfig;    @Autowired    private WebSocketHandler webSocketHandler;    @Override    protected void initChannel(Channel channel) throws Exception {        ChannelPipeline pipeline = channel.pipeline();        //对http协议的支持        pipeline.addLast(new HttpServerCodec());        //对大数据流的支持        pipeline.addLast(new ChunkedWriteHandler());        //post请求分为3部分。request line、request header、body        //HttpObjectAggregator将多个信息转化为单一的request或者response对象        pipeline.addLast(new HttpObjectAggregator(8000));        //将http协议升级为ws协议，websocket的支持        pipeline.addLast(new WebSocketServerProtocolHandler(nettyConfig.getPath()));        pipeline.addLast(webSocketHandler);    }}

@Component@ChannelHandler.Sharable  //设置通道共享public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {    private List<Channel> channels=new ArrayList<>();    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        channels.add(ctx.channel());        System.out.println("有新的连接了...");    }    @Override    public void channelInactive(ChannelHandlerContext ctx) throws Exception {        channels.remove(ctx.channel());        System.out.println("连接下线了");    }    @Override    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {        String text = textWebSocketFrame.text();        Channel currentChannel = channelHandlerContext.channel();        for (Channel channel:channels){            //自己不给自己发消息            if(!channel.equals(currentChannel)){                channel.writeAndFlush(new TextWebSocketFrame(text));            }        }    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        cause.printStackTrace();        Channel channel = ctx.channel();        channels.remove(channel);    }}

3. 前端js代码

$(function () {    //这里需要注意的是，prompt有两个参数，前面是提示的话，后面是当对话框出来后，在对话框里的默认值    var username = "";    while (true) {        //弹出一个输入框，输入一段文字，可以提交        username = prompt("请输入您的名字", ""); //将输入的内容赋给变量 name ，        if (username.trim() === "")//如果返回的有内容        {            alert("名称不能输入空")        } else {            $("#username").text(username);            break;        }    }    var ws = new WebSocket("ws://localhost:8081/chatService");    ws.onopen = function () {        console.log("连接成功.")    };    ws.onmessage = function (evt) {        showMessage(evt.data);    };    ws.onclose = function (){        console.log("连接关闭")    };    ws.onerror = function (){        console.log("连接异常")    };    function showMessage(message) {        // 张三:你好        var str = message.split(":");        $("#msg_list").append('<li }>\n' +            '                                  <div >\n' +            '                                    <img  width="30" height="30" src="/img/user.png">\n' +            '                                    <div>\n' +            '                                        <div >'+str[0]+'</div>\n' +            '                                        <div >'+str[1]+'</div>\n' +            '                                    </div>                       \n' +            '                                   </div>\n' +            '                              </li>');        // 置底        setBottom();    }    $('#my_test').bind({        focus: function (event) {            event.stopPropagation();            $('#my_test').val('');            $('.arrow_box').hide()        },        keydown: function (event) {            event.stopPropagation();            if (event.keyCode === 13) {                if ($('#my_test').val().trim() === '') {                    this.blur();                    $('.arrow_box').show();                    setTimeout(this.focus(),1000);                } else {                    $('.arrow_box').hide();                    //发送消息                    sendMsg();                    this.blur();                    setTimeout(this.focus())                }            }        }    });    $('#send').on('click', function (event) {        event.stopPropagation();        if ($('#my_test').val().trim() === '') {            $('.arrow_box').show()        } else {            sendMsg();        }    });    function sendMsg() {        var message = $("#my_test").val();        $("#msg_list").append('<li }>\n' +            '                                  <div >\n' +            '                                      <div >'+message+'</div>\n' +            '                                  </div>\n' +            '                              </li>');        $("#my_test").val('');        //发送消息        message = username + ":" + message;        ws.send(message);        // 置底        setBottom();    }    // 置底    function setBottom() {        // 发送消息后滚动到底部        var container = $('.m-message');        var scroll = $('#msg_list');        container.animate({            scrollTop: scroll[0].scrollHeight - container[0].clientHeight + container.scrollTop() + 100        });    }});