Netty入门之WebSocket初体验

作者: Hey_Shaw | 来源:发表于2018-06-19 22:39 被阅读31次

什么是Netty

高性能事件驱动、异步非阻塞；基于Netty可以建立高性能的HTTP服务器。
基于NIO的客户端，服务器端编程框架；可以快速简单开发网络应用，比如TCP/UDP的socket网络开发。
稳定性和伸缩性

Netty使用场景

高性能领域
多线程并发领域
异步通信领域

BIO通信

一个线程负责连接
一请求一应答
缺乏弹性伸缩能力

BIO通信模型

伪异步IO通信

线程池负责连接
M请求N应答
线程池阻塞

伪异步IO通信模型

NIO通信

缓冲区Buffer
通道Channel
多路复用器Selector

AIO通信

连接注册读写事件和回调函数
读写方法异步
主动通知程序

四种IO对比

客户端个数
IO类型（按顺序依次为：阻塞同步IO、阻塞同步IO、非阻塞同步IO、非阻塞异步IO）
API使用难度
调试难度
可靠性
吞吐量

原生NIO的缺陷

类库和API繁杂
工作量和难度大
入门门槛高
JDK NIO存在bug

Netty的优势

API简单
性能高
入门门槛低
成熟、稳定

WebSocket入门

什么是WebSocket？

H5协议规范
握手机制
解决客户端与服务端实时通信而产生的技术

WebSocket的优点

节省通信开销
服务器主动传送数据给客户端
实时通信

WebSocket建立连接

客户端发起握手请求
服务端响应请求
连接建立

WebSocket生命周期

打开事件
消息事件
错误事件
关闭事件

WebSocket连接关闭

服务器关闭底层TCP连接
客户端发起TCP Close

Netty实现WebSocket通信案例

服务端

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 存储整个工程的全局配置
 * @author liuyazhuang
 *
 */
public class NettyConfig {
    
    /**
     * 存储每一个客户端接入进来时的channel对象
     */
    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

import java.util.Date;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;

/**
 * 接收/处理/响应客户端websocket请求的核心业务处理类
 * @author liuyazhuang
 *
 */
public class MyWebSocketHandler extends SimpleChannelInboundHandler<Object> {
    
    private WebSocketServerHandshaker handshaker;
    private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket";

    // 客户端与服务端创建连接的时候调用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        NettyConfig.group.add(ctx.channel());
        System.out.println("客户端与服务端连接开启...");
    }

    // 客户端与服务端断开连接的时候调用
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        NettyConfig.group.remove(ctx.channel());
        System.out.println("客户端与服务端连接关闭...");
    }

    // 服务端接收客户端发送过来的数据结束之后调用
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
    }

    // 工程出现异常的时候调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    // 服务端处理客户端websocket请求的核心方法
    @Override
    protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception {
        // 处理客户端向服务端发起http握手请求的业务
        if (msg instanceof FullHttpRequest) {
            handHttpRequest(context,  (FullHttpRequest)msg);
        }else if (msg instanceof WebSocketFrame) { //处理websocket连接业务
            handWebsocketFrame(context, (WebSocketFrame)msg);
        }
    }
    
    /**
     * 处理客户端与服务端之前的websocket业务
     * @param ctx
     * @param frame
     */
    private void handWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){
        // 判断是否是关闭websocket的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
        }
        // 判断是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        
        // 判断是否是二进制消息，如果是二进制消息，抛出异常
        if( ! (frame instanceof TextWebSocketFrame) ){
            System.out.println("目前我们不支持二进制消息");
            throw new RuntimeException("【"+this.getClass().getName()+"】不支持消息");
        }
        // 返回应答消息
        // 获取客户端向服务端发送的消息
        String request = ((TextWebSocketFrame) frame).text();
        System.out.println("服务端收到客户端的消息====>>>" + request);
        TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() 
                                                                                                + ctx.channel().id() 
                                                                                                + " ===>>> " 
                                                                                                + request);
        // 群发，服务端向每个连接上来的客户端群发消息
        NettyConfig.group.writeAndFlush(tws);
    }
    /**
     * 处理客户端向服务端发起http握手请求的业务
     * @param ctx
     * @param req
     */
    private void handHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req){
        if (!req.getDecoderResult().isSuccess() 
                || ! ("websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req, 
                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                WEB_SOCKET_URL, null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
        }else{
            handshaker.handshake(ctx.channel(), req);
        }
    }
    
    /**
     * 服务端向客户端响应消息
     * @param ctx
     * @param req
     * @param res
     */
    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req,
            DefaultFullHttpResponse res){
        if (res.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
        // 服务端向客户端发送数据
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (res.getStatus().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }
}

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;

/**
 * 初始化连接时候的各个组件
 * @author liuyazhuang
 *
 */
public class MyWebSocketChannelHandler extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel e) throws Exception {
        e.pipeline().addLast("http-codec", new HttpServerCodec());
        e.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
        e.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
        e.pipeline().addLast("handler", new MyWebSocketHandler());
    }

}

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * 程序的入口，负责启动应用
 * @author liuyazhuang
 *
 */
public class Main {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workGroup);
            b.channel(NioServerSocketChannel.class);
            b.childHandler(new MyWebSocketChannelHandler());
            System.out.println("服务端开启等待客户端连接....");
            Channel ch = b.bind(8888).sync().channel();
            ch.closeFuture().sync();
            
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            //优雅的退出程序
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

客户端

<html>
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset = utf-8"/>
        <title>WebSocket客户端</title>
    <script type="text/javascript">
        var socket;
        if(!window.WebSocket){
            window.WebSocket = window.MozWebSocket;
        }

        if(window.WebSocket){
            socket = new WebSocket("ws://localhost:8888/websocket");
            socket.onmessage = function(event){
                var ta = document.getElementById('responseContent');
                ta.value += event.data + "\r\n";
            };

            socket.onopen = function(event){
                var ta = document.getElementById('responseContent');
                ta.value = "你当前的浏览器支持WebSocket,请进行后续操作\r\n";
            };

            socket.onclose = function(event){
                var ta = document.getElementById('responseContent');
                ta.value = "";
                ta.value = "WebSocket连接已经关闭\r\n";
            };
        }else{
            alert("您的浏览器不支持WebSocket");
        }


        function send(message){
            if(!window.WebSocket){
                return;
            }
            if(socket.readyState == WebSocket.OPEN){
                socket.send(message);
            }else{
                alert("WebSocket连接没有建立成功！！");
            }
        }
    </script>
    </head>
    <body>
        <form onSubmit="return false;">
            <input type = "text" name = "message" value = ""/>
            <br/><br/>
            <input type = "button" value = "发送WebSocket请求消息" onClick = "send(this.form.message.value)"/>
            <hr color="red"/>
            <h2>客户端接收到服务端返回的应答消息</h2>
            <textarea id = "responseContent" style = "width:1024px; height:300px"></textarea>
        </form>
    </body>
</html>

Netty入门之WebSocket初体验

什么是Netty

Netty使用场景

BIO通信

伪异步IO通信

NIO通信

AIO通信

四种IO对比

原生NIO的缺陷

Netty的优势

WebSocket入门

什么是WebSocket？

WebSocket的优点

WebSocket建立连接

WebSocket生命周期

WebSocket连接关闭

Netty实现WebSocket通信案例

服务端

客户端

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读