Netty心跳检测和基于Websocket协议的服务端开发

心跳检测

在 TCP 长连接中,Netty服务端感知客户端(或是客户端感知服务端)断开连接的其中一个方法是handler的channelInactive。但可能会有一些情况,比如线路出现问题,让客户端和服务端虽然是连接状态但事实无法通信。这种情况就通过心跳检测对方是否有响应,心跳检测(心跳包的发送)需要编程人员在客户端和服务端层面实现,而Netty中提供了IdleStateHandler用于处理通道空闲的情况(在心跳机制启用的情况下就代表连接断开)。

//第一个参数没有读事件的时间,第二个参数没有写事件的时间
//第三个参数既没有读事件也没有写事件的时间
//设置为0代表不限制
//满足任意一个则触发IdleStateEvent,可以在userEventTriggered中处理
pipeline.addLast(new IdleStateHandler(10,10,6, TimeUnit.SECONDS));

handler中:

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if(evt instanceof IdleStateEvent){
        IdleStateEvent event = (IdleStateEvent)evt;
        String eventType = null;
        switch (event.state()){
            case ALL_IDLE:
                eventType = "读写空闲";
                break;
            case READER_IDLE:
                eventType = "读空闲";
                break;
            case WRITER_IDLE:
                eventType = "写空闲";
                break;
        }
        System.out.println(ctx.channel().remoteAddress()+"--超时时间--"+eventType);
        //服务器可以断开该channel连接
        //ctx.channel.close();
    }
}

基于Websocket的服务端开发

首先Websocket协议是基于Http协议的,它借用了HTTP协议来完成一部分握手,所以连接的请求看起来可能是这样的:

GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13
Origin: http://example.com

所以首先要给pipeline加一个HttpServerCodec

pipeline.addLast("MyHttpServerCodec",new HttpServerCodec());

然后是Http相关的handler

pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(8192));

然后是websocket核心handler:

//将http协议升级为websocket协议,参数代表请求的uri
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));

然后写一个处理消息的处理器:

public class TextWebsocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("服务端收到消息:"+msg.text());
        //回复消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间【"+ LocalDateTime.now()+"】:"+msg.text()));
    }

    //连接建立时,handlerAdded被调用
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("handlerAdded被调用"+ctx.channel().id().asLongText());
    }

    //连接断开时,handlerRemoved被调用
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("handlerRemoved被调用"+ctx.channel().id().asLongText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常发生"+cause.getMessage());
        ctx.close();//关闭连接
    }
}

这个处理器中的channelRead0可以将客户端发来的消息加上服务器时间返回给客户端

服务器启动类总体看起来是这样的:

public class WebsocketServer {
    public void bind(int port) throws Exception{
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new ChunkedWriteHandler());
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
                            pipeline.addLast(new TextWebsocketFrameHandler());
                        }
            });
            System.out.println("---------服务器正在启动---------");
            ChannelFuture future = serverBootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws Exception {
        int port = 7000;
        new WebsocketServer().bind(port);
    }
}

编写网页进行测试:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>

<body>
    <form onsubmit="return false">
        <textarea name="message" style="height: 300px;width: 300px"></textarea>
        <input type="button" value="发送消息" onclick="send(this.form.message.value)">
        <textarea id="responseText" style="height: 300px;width: 300px"></textarea>
        <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">
    </form>
</body>
<script>
    let socket;
    if(window.WebSocket){
        let rt = document.getElementById("responseText");
        socket = new WebSocket("ws://localhost:7000/ws");
        //收到消息
        socket.onmessage = function (ev) {
            rt.value += "\n" + ev.data;
        }
        //连接开启
        socket.onopen = function (ev) {
            rt.value = "--------连接开启了---------";
        }
        socket.onclose = function (ev) {
            rt.value += "\n" + "-------连接关闭了-------";
        }
    }else {
        alert("当前浏览器不支持websocket");
    }
    //发送消息
    function send(message) {
        if(!socket){//先判断socket是否创建好
            return;
        }
        if(socket.readyState == WebSocket.OPEN){
            //发送消息
            socket.send(message);
        }else{
            alert("连接没有开启");
        }
    }
</script>
</html>

测试:

Netty心跳检测和基于Websocket协议的服务端开发

原创文章,作者:彭晨涛,如若转载,请注明出处:https://www.codetool.top/article/netty%e5%bf%83%e8%b7%b3%e6%a3%80%e6%b5%8b%e5%92%8c%e5%9f%ba%e4%ba%8ewebsocket%e5%8d%8f%e8%ae%ae%e7%9a%84%e6%9c%8d%e5%8a%a1%e7%ab%af%e5%bc%80%e5%8f%91/

发表评论

电子邮件地址不会被公开。