Netty -Netty心跳检测机制案例,Netty通过WebSocket编程实现服务器和客户端长链接

84 阅读 0 评论 56 点赞

我是靠谱客的博主爱撒娇小鸭子，最近开发中收集的这篇文章主要介绍Netty -Netty心跳检测机制案例,Netty通过WebSocket编程实现服务器和客户端长链接，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Netty心跳检测机制案例

案例要求
- 编写一个Netty心跳检测机制案例，当服务器超过3秒没有读时，就提示读空闲
- 当服务器超过5秒没有写操作时，提示写空闲
- 服务器超过7秒没有读或者写操作时，就提示读写空闲

代码

HeartBeatServer

package com.jl.java.web.heartbeat;

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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @author jiangl
 * @version 1.0
 * @date 2021/5/25 10:32
 */
public class HeartBeatServer {

    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();

            bootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childOption(ChannelOption.SO_KEEPALIVE,true)
                    //handler是bossGroup的事件
                    .handler(new LoggingHandler(LogLevel.INFO))
                    //childHandler是工作线程组 workerGroup的事件
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            /*
                             * 说明
                             * 1.IdleStateHandler 是Netty提供的处理空闲状态的处理器
                             * 2.long readerIdleTime：表示多长时间没有读，就会发送一个心跳检测包检查是否连接
                             * 3.long writerIdleTime：表示多长时间没有写，就会发送一个心跳检测包检查是否连接
                             * 4.long allIdleTime：表示多长时间没有读写，就会发送一个心跳检测包检查是否连接
                             * 5.文档说明
                             * Triggers an IdleStateEvent when a Channel has not performed read, write, or both operation for a while.
                             * 当通过在一段时间内没有发生读，写或者读写操作时，会触发一个IdleStateEvent（空闲事件）
                             * 6. 当IdleStateEvent触发后，就会传递给管道的下一个Handler去处理，通过触发下一个handler的userEventTriggered时间，在
                             * 该方法处理IdleStateEvent（读空闲，写空闲，读写空闲）
                             */
                            pipeline.addLast(new IdleStateHandler(3,5,7, TimeUnit.SECONDS));
                            //解码器
                            pipeline.addLast("decode",new StringDecoder());
                            //编码器
                            pipeline.addLast("encode",new StringEncoder());
                            //自定义的Handler
                            pipeline.addLast(new HeartBeatServerHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(7000).sync();
            //将关闭事件同步
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

HeartBeatServerHandler

package com.jl.java.web.heartbeat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @author jiangl
 * @version 1.0
 * @date 2021/5/25 10:32
 */
public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if( evt instanceof IdleStateEvent){
            //将Evt向下转型
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            String eventType = null;
            switch(idleStateEvent.state() ){
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            System.out.println(ctx.channel().remoteAddress()+"----超时时间----"+eventType);
            System.out.println("服务器做响应的处理...");
        }
    }
}

Netty通过WebSocket编程实现服务器和客户端长链接

案例要求
- HTTP协议是无状态的，浏览器和服务器间的请求响应一次，下一次会重新创建连接
- 要求：实现基于websocket的长链接的全双工的交互
- 改变http协议多次请求的约束，实现长链接，服务器可以发送消息给浏览器
- 客户端浏览器和服务器端会相互感知，比如服务器关闭了，浏览器会感知，同样浏览器关闭了，服务器会感知

代码

MyServer

package com.jl.java.web.websocket;

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.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

/**
 * @author jiangl
 * @version 1.0
 * @date 2021/5/25 21:54
 */
public class MyServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,128)
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //因为基于http协议，使用http的编码和解码器
                            pipeline.addLast("decoder",new HttpServerCodec());
                            //是以块方式写，添加ChunkedWriteHandler处理器
                            pipeline.addLast("encoder",new ChunkedWriteHandler());
                            /*
                             * HttpObjectAggregator说明
                             * 1.http数据在传输过程中是分段，HttpObjectAggregator，就是可以将多个段聚合
                             * 2.当浏览器发送大量数据时，就会发出多次http请求
                             */
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            /*
                             * WebSocketServerProtocolHandler说明
                             * 1.对应WebSocket，它的数据是以 帧（Frame）形式传递
                             * 2.WebSocketFrame 有六个子类
                             * 3.浏览器请求时，ws://localhost:7000/hello
                             * 4.WebSocketServerProtocolHandler 核心功能将Http协议升级为ws协议 即ws协议（保持长链接）
                             */
                            pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
                            //添加自定义的handler事件，专门处理浏览器请求，处理业务逻辑
                            pipeline.addLast(new MyTextWebSocketFrameHandler());

                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

MyTextWebSocketFrameHandler

package com.jl.java.web.websocket;

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

import java.time.LocalDateTime;

/**
 * @author jiangl
 * @version 1.0
 * @date 2021/5/26 10:10
 */
public class MyTextWebSocketFrameHandler 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()));
    }

    /**
     * 当web客户端连接后，触发方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        //id表示唯一的值，LongText 是唯一的 shortText不是唯一的
        System.out.println("handlerAdded 被调用："+ctx.channel().id().asLongText());
        System.out.println("handlerAdded 被调用："+ctx.channel().id().asShortText());
    }

    @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();//关闭连接
    }
}

html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <script type="application/javascript">
        var socket;
        if(window.WebSocket){
            socket = new WebSocket("ws://localhost:8080/hello");
            //相当于channelRead,ev接收服务端返回的消息
            socket.onmessage = function (ev){
                var rt = document.getElementById("responseText");
                rt.value = rt.value + "n" +ev.data;
            }
            //相当于连接开启
            socket.onopen = function (ev){
                var rt = document.getElementById("responseText");
                rt.value = "连接开始...." + "n";
            }
            //连接关闭
            socket.onclose = function(ev){
                var rt = document.getElementById("responseText");
                rt.value = rt.value + "n" +"连接关闭....";
            }
        }else{
            alert("当前浏览器不支持webSocket");
        }

        //发送消息
        function send(message){
            if(!window.socket){
                return;
            }
            if(socket.readyState == WebSocket.OPEN){
                //通过Socket发送消息
                socket.send(message);
            }else{
                alert("连接没有开启")
            }
        }

    </script>
    <form onsubmit="return false">
        <textarea id="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>
</html>