Netty实现WebSocket

场景

由于Http协议是无状态的,每一次请求只能响应一次,下次请求需要重新连接。

如果客户端请求一个服务端资源,需要实时监服务端执行状态(比如导出大数据量时需要前端监控导出状态),这个时候不断请求连接浪费资源。可以通过WebSocket建立一个长连接,实现客户端与服务端双向交流。

使用Netty实现浏览器与服务端建立WebSocket连接,互相监控状态,客户端发送消息服务端回写。

服务端状态及消息发送及回显:

Netty实现WebSocket

服务端读取浏览器消息并监控页面状态

Netty实现WebSocket

实现

服务端

服务端需要添加多个Netty框架的Handler,其中使用WebSocketServerProtocolHandler("/hello")将http协议升级为WebSocket协议,升级指定的uri需要与浏览器请求地址保持一致。

package others.netty.webSocket;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
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.stream.ChunkedWriteHandler;

/**
 * Netty实现WebSocket实例
 * Http连接是无状态的,且每一个请求只会响应一次,下次需要重新连接。服务端不可主动向客户端发送消息
 * 使用WebSocket实现一个客户端与服务端可互相通信的偿连接
 *
 * @author makeDoBetter
 * @version 1.0
 * @date 2021/4/29 15:54
 * @since JDK 1.8
 */
public class Server {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //基于http协议,因此添加http编解码器
                            pipeline.addLast(new HttpServerCodec());
                            //提供大文件写的处理器,尤其适用于大文件写,方便管理状态,不需要用户过分关心
                            //一个{@link ChannelHandler},它增加了对写入大型数据流的支持既不花费大量内存
                            // 也不获取{@link OutOfMemoryError}。
                            // 大型数据流(例如文件传输)需要在{@link ChannelHandler}实现中进行复杂的状态管理。
                            // {@link ChunkedWriteHandler}管理如此复杂的状态以便您可以毫无困难地发送大量数据流。
                            pipeline.addLast(new ChunkedWriteHandler());
                            //将http协议下分段传输的数据聚合到一起,用于响应请求是需要添加在 HttpServerCodec()之后
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            //将服务器协议升级为WebSocket协议保持长连接 处理握手及帧的传递
                            //升级协议是通过修改状态码实现的 200升级为101
                            //WebSocket 长连接消息传递是通过帧的形式进行传递的
                            //帧 继承抽象类 WebSocketFrame 有六个子类 帧的处理由管道中下一个handler进行处理
                            //WebSocket请求形式 :ws://localhost:1234/hello
                            pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
                            pipeline.addLast(new FrameHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(1234).sync();
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        System.out.println("服务端启动");
                    } else {
                        System.out.println("服务端启动失败");
                    }
                }
            });
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

自定义处理器
自定义处理器实现浏览器消息的接收及回写,以及实现其他连接及离线事件的监控。

package others.netty.webSocket;

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

import java.time.LocalDateTime;

/**
 * WebSocket 长连接下 文本帧的处理器
 * 实现浏览器发送文本回写
 * 浏览器连接状态监控
 *
 * @author makeDoBetter
 * @version 1.0
 * @date 2021/4/29 16:30
 * @since JDK 1.8
 */
public class FrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //使用msg.text()获得帧中文本
        System.out.println(msg.text());
        //回写,需要封装成TextWebSocketFrame 对象写入到通道中
        ctx.channel().writeAndFlush(new TextWebSocketFrame("【服务端】" + LocalDateTime.now() + msg.text()));
    }

    /**
     * 出现异常的处理 打印报错日志
     *
     * @param ctx   the ctx
     * @param cause the cause
     * @throws Exception the Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
    }

    /**
     * 监控浏览器上线
     *
     * @param ctx the ctx
     * @throws Exception the Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().id().asShortText() + "连接");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().id().asShortText() + "断开连接");
    }
}

客户端

客户端思路:

  • 如果浏览器支持WebSocket,创建一个WebSocket连接,后使用socket变量监控各个事件,并给出相应事件的发生;
  • 发送消息时先校验是否允许WebSocket,并在socket.readyState == WebSocket.OPEN状态下进行消息发送。
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var socket;
    if (window.WebSocket) {
        socket = new WebSocket("ws://localhost:1234/hello");
        //消息获得事件
        socket.onmessage = function (ev) {
            alert("收到消息");
            var id = document.getElementById(‘getMessage‘);
            id.value = id.value + ‘\n‘ + ev.data;
        };
        socket.onopen = function (ev) {
            var id = document.getElementById(‘getMessage‘);
            id.value = ‘连接服务器成功‘;
        };
        socket.onclose = function (ev) {
            var id = document.getElementById(‘getMessage‘);
            id.value = id.value + ‘\n‘ + ‘服务器连接关闭‘;
        }
    } else {
        alert("浏览器不支持WebSocket");
    }

    function send(message) {
        if (!window.socket) {
            return;
        }
        if (socket.readyState == WebSocket.OPEN) {
            socket.send(message);
        }
    }
</script>
<form onsubmit="return false">
    <textarea name="textarea1" type="text" style="width: 300px; height: 400px"></textarea>
    <input type="button" onclick="send(this.form.textarea1.value)" value="发送">
    <textarea id="getMessage" type="text" style="width: 300px; height: 400px"></textarea>
    <input type="button" onclick="document.getElementById(‘getMessage‘).value=‘‘" value="清空">
</form>

</body>
</html>

Netty实现WebSocket

上一篇:Flume


下一篇:node.js学习