16-基于Netty开发WebSocket服务器与浏览器实现长链接

package com.dance.netty.netty.websocket;

import com.dance.netty.netty.heartbeat.NettyServerHertBeat;
import com.dance.netty.netty.heartbeat.NettyServerIdleStateHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

public class WebSocketServer {
    private final int port;

    public WebSocketServer(int port) {
        this.port = port;
    }

    public void run() throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new LoggingHandler(LogLevel.INFO))  // 在BossGroup中增加一个日志处理器 日志级别为INFO
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 因为是基于Http协议的所以采用Http的编解码器
                            pipeline.addLast(new HttpServerCodec());
                            // 是以块的方式写, 添加ChunkedWriteHandler(分块写入处理程序)
                            pipeline.addLast(new ChunkedWriteHandler());
                            /*
                             * http 数据在传输过程中是分段的 http Object aggregator 就是可以将多个段聚合
                             * 这就是为什么 当浏览器发送大量数据时, 就会出现多次http请求
                             * 参数为 : 最大内容长度
                             */
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            /*
                             * 对应WebSocket 他的数据时以桢(frame) 形式传递
                             * 可以看到WebSocketFrame下面有6个子类
                             * 浏览器请求时: ws://localhost:7000/xxx 请求的url
                             * 核心功能是将http协议升级为ws协议 保持长链接
                             */
                            pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
                            // 自定义Handler, 处理业务逻辑
                            pipeline.addLast(new WebSocketTextFrameHandler());
                        }
                    });
            System.out.println("netty server is starter......");
            ChannelFuture sync = serverBootstrap.bind(port).sync();
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

    public static void main(String[] args) throws InterruptedException {
        new WebSocketServer(7000).run();
    }
}

package com.dance.netty.netty.websocket;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import java.time.LocalDateTime;

/**
 * TextWebSocketFrame 表示一个文本桢
 */
public class WebSocketTextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("[服务器] : 收到消息 -> " + msg.text());
        // 回复浏览器
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->"+ msg.text()));
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // id 表示唯一的值 LongText是唯一的
        System.out.println("handlerAdded 被调用:" + ctx.channel().id().asLongText());
        // shortText 可能会重复
        System.out.println("handlerAdded 被调用:" + ctx.channel().id().asShortText());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        // id 表示唯一的值 LongText是唯一的
        System.out.println("handlerRemoved 被调用:" + ctx.channel().id().asLongText());
        // shortText 可能会重复
        System.out.println("handlerRemoved 被调用:" + ctx.channel().id().asShortText());
    }

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

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var socket;
    // 判断当前浏览器是否支持WebSocket
    if(window.WebSocket){
        socket = new WebSocket("ws://localhost:7000/hello");
        // 相当于ChannelRead ev就是消息回送
        socket.onmessage = function (ev){
            console.log(ev)
            let textArea = document.getElementById("responseText")
            textArea.value = textArea.value + "\n" + ev.data
        }
        // 相当于连接开启 ChannelAdded ev
        socket.onopen = function (ev) {
            console.log(ev)
            let textArea = document.getElementById("responseText")
            textArea.value = "连接开启了"
        }
        // 相当于连接关闭 ChannelRemove ev
        socket.onclose = function (ev) {
            console.log(ev)
            let textArea = document.getElementById("responseText")
            textArea.value = textArea.value + '\n' + "连接关闭了"
        }
    }else{
        alert("当前浏览器不支持WebSocket")
    }
    function send(msg) {
        if(!window.socket){
            return;
        }
        if(socket.readyState === WebSocket.OPEN){
            // 通过WebSocket发送消息
            socket.send(msg)
        }else{
            alert('连接没有开启')
        }
    }
</script>
<form onsubmit="return false">
    <textarea name="message" style="width: 300px; height: 300px;"></textarea>
    <input type="button" value="发送消息" onclick="send(this.form.message.value)">
    <textarea id="responseText" style="width: 300px; height: 300px;"></textarea>
    <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">

</form>
<script>


</script>

</body>
</html>

netty server is starter......
一月 16, 2022 11:19:27 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0xb0f42cfd] REGISTERED
一月 16, 2022 11:19:27 下午 io.netty.handler.logging.LoggingHandler bind
信息: [id: 0xb0f42cfd] BIND: 0.0.0.0/0.0.0.0:7000
一月 16, 2022 11:19:27 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0xb0f42cfd, L:/0:0:0:0:0:0:0:0:7000] ACTIVE

一月 16, 2022 11:20:46 下午 io.netty.handler.logging.LoggingHandler channelRead
信息: [id: 0xb0f42cfd, L:/0:0:0:0:0:0:0:0:7000] READ: [id: 0x9b6bccb3, L:/0:0:0:0:0:0:0:1:7000 - R:/0:0:0:0:0:0:0:1:55639]
一月 16, 2022 11:20:46 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
信息: [id: 0xb0f42cfd, L:/0:0:0:0:0:0:0:0:7000] READ COMPLETE
handlerAdded 被调用:005056fffec00008-00006534-00000002-37865b9aaa734014-9b6bccb3
handlerAdded 被调用:9b6bccb3

[服务器] : 收到消息 -> hi netty

handlerRemoved 被调用:005056fffec00008-00006534-00000002-37865b9aaa734014-9b6bccb3
handlerRemoved 被调用:9b6bccb3