基于Netty开发Websocket服务端

码客说

发布于 2019-10-22 17:09:51

1.8K0

发布于 2019-10-22 17:09:51

文章被收录于专栏：码客码客码客

Netty简介

那么Netty到底是何方神圣？

用一句简单的话来说就是：Netty封装了JDK的NIO，让你用得更爽，你不用再写一大堆复杂的代码了。
用官方正式的话来说就是：Netty是一个异步事件驱动的网络应用框架，用于快速开发可维护的高性能服务器和客户端。

下面是我总结的使用Netty不使用JDK原生NIO的原因

使用JDK自带的NIO需要了解太多的概念，编程复杂
Netty底层IO模型随意切换，而这一切只需要做微小的改动，改改参数，Netty可以直接从NIO模型变身为IO模型
Netty自带的拆包解包，异常检测等机制让你从NIO的繁重细节中脱离出来，让你只需要关心业务逻辑
Netty解决了JDK的很多包括空轮询在内的bug
Netty底层对线程，selector做了很多细小的优化，精心设计的reactor线程模型做到非常高效的并发处理
自带各种协议栈让你处理任何一种通用协议都几乎不用亲自动手
Netty已经历各大rpc框架，消息中间件，分布式通信中间件线上的广泛验证，健壮性无比强大

Websocket服务端

引入Netty

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>5.0.0.Alpha1</version>
</dependency>

初始化代码

private static void initWebScoket() {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup wokerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap
            .group(bossGroup, wokerGroup)
            .channel(NioServerSocketChannel.class)
            .handler(new LoggingHandler(LogLevel.INFO))
            .option(ChannelOption.SO_KEEPALIVE, true)
            .childHandler(new ChannelInitializer<SocketChannel>(){
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
                    pipeline.addLast(new HttpServerCodec());
                    //以块的方式来写的处理器
                    pipeline.addLast(new ChunkedWriteHandler());
                    //netty是基于分段请求的，HttpObjectAggregator的作用是将请求分段再聚合,参数是聚合字节的最大长度
                    pipeline.addLast(new HttpObjectAggregator(1024 * 1024 * 1024));
                    //ws://localhost:8899/ws
                    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
                    //websocket定义了传递数据的6中frame类型
                    pipeline.addLast(new WebSocketHandle());
                }
            });

        ChannelFuture channelFuture = serverBootstrap
            .bind(new InetSocketAddress(8899))
            .sync();

        channelFuture
            .channel()
            .closeFuture()
            .sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        bossGroup.shutdownGracefully();
        wokerGroup.shutdownGracefully();
    }
}

bossGroup对应接受新连接线程，主要负责创建新连接
wokerGroup对应负责读取数据的线程，主要用于读取数据以及业务逻辑处理

WebSocketHandle

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

public class WebSocketHandle extends SimpleChannelInboundHandler<Object> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("用户连接");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("用户断开");
    }

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof TextWebSocketFrame) {
            String receive_msg = ((TextWebSocketFrame) msg).text();
            System.out.println("收到消息：" + receive_msg);
            String send_msg = "服务器返回：" + receive_msg;
            System.out.println("ctx.channel.id:"+ctx.channel().id());
            //服务端返回消息
            ctx.channel().writeAndFlush(new TextWebSocketFrame(send_msg));
        } else if (msg instanceof BinaryWebSocketFrame) {
            System.out.println("收到二进制消息：" + ((BinaryWebSocketFrame) msg).content().readableBytes());
            BinaryWebSocketFrame binaryWebSocketFrame = new BinaryWebSocketFrame(Unpooled.buffer().writeBytes("xxx".getBytes()));
            ctx.channel().writeAndFlush(binaryWebSocketFrame);
        }
    }

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

Websocket客户端

这样在JS中既可以这样连接

var socket = new WebSocket("ws://127.0.0.1:8899/ws");
socket.onopen = function () {
    
};
socket.onmessage = function (evt) {
    
};
socket.onclose = function (evt) {
    
}
socket.onerror = function (evt) {
    
}

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2019-02-26，如有侵权请联系 cloudcommunity@tencent.com 删除

java