Netty之WebSocket协议应用开发

public class WSServer {

    public static void main(String[] args) {
        new WSServer().start();
    }

    private void start() {
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap server = new ServerBootstrap();
            server.group(boss,worker)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //将请求和应答消息编码或解码成http消息
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            //支持异步发送大的码流
                            pipeline.addLast(new ChunkedWriteHandler());
                            pipeline.addLast(new WSServerHandler());
                        }

                    })
                    .childOption(ChannelOption.SO_KEEPALIVE,true);

            ChannelFuture sync = server.bind(8888).sync();

            System.out.println("WebSocket服务端已启动，端口号为8888");

            sync.channel().closeFuture().sync();

        } catch (Exception e) {

        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

public class WSServerHandler extends SimpleChannelInboundHandler<Object> {


    private static final Log logger = LogFactory.getLog(WSServerHandler.class);


    private WebSocketServerHandshaker handshaker;

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
        //如果是握手请求
        if (msg instanceof FullHttpRequest) {
            handlerHttpRequest(ctx,(FullHttpRequest)msg);
        } else if (msg instanceof WebSocketFrame) {//如果是WebSocket消息接受
            handlerWebSocketFrame(ctx,(WebSocketFrame)msg);
        }
    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

        //判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return ;
        }

        //判断是否是ping指令
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        //只支持文本信息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(frame.getClass().getName()+" frame type not support ");
        }


        String text = ((TextWebSocketFrame) frame).text();
        logger.info("服务端接收到客户端信息为："+text);
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        ctx.writeAndFlush(new TextWebSocketFrame("欢迎使用Netty WebSocket服务，目前时间为："+ format.format(new Date())));
    }

    private void handlerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {


        //如果http解码失败，返回http异常
        //判断是否是WebSocket握手请求
        if (!req.decoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sentHttpResponse(ctx,req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        //构造握手响应返回
        WebSocketServerHandshakerFactory wsFactory =
                new WebSocketServerHandshakerFactory("ws://localhost:8888", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(),req);
        }
    }

    private void sentHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
        if (res.status().code() != 200) {
            ByteBuf byteBuf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(byteBuf);
            byteBuf.release();
            setContentLength(res,res.content().readableBytes());
        }

        //如果是非keep-alive连接，关闭连接
        ChannelFuture future = ctx.channel().writeAndFlush(res);
        if (!isKeepAlive(req) || res.status().code()!=200) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

}