Netty实现websocket示例-蒲公英云

这里构建一个websocket的服务端，客户端利用html页面来模拟，网上有很多netty实现websocket的示例，但是都不完整，也没有演示效果，这里给出一个完整的，既有服务端的实现，也有页面模拟客户端聊天功能的实现。

工程结构：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70

pom.xml

io.netty
netty-all
4.1.27.Final

NettyConfig.java

package com.xxx.netty.websocket;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
public class NettyConfig {
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

WebsocketChannelHandler.java

package com.xxx.netty.websocket;
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.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
public class WebSocketChannelHandler extends SimpleChannelInboundHandler<Object> {
    private WebSocketServerHandshaker handshaker;
    private static final String web_socket_url="ws://127.0.0.1:8888/websocket";
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof FullHttpMessage) {
            handleHttpRequest(ctx,(FullHttpRequest)msg);
        }else if(msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx,(WebSocketFrame)msg);
        }
    }
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if(frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
        }
        if(frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if(!(frame instanceof TextWebSocketFrame)) {
            System.out.println("目前我们不支持二进制消息");
            new RuntimeException("["+this.getClass().getName()+"]不支持消息");
        }
        String request = ((TextWebSocketFrame)frame).text();
        System.out.println("服务端接收到消息==>"+request);
        TextWebSocketFrame res = new TextWebSocketFrame(new Date().toString()+"-"+ctx.channel().id()+"===>"+request);
        NettyConfig.group.writeAndFlush(res);
    }
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        if(!request.decoderResult().isSuccess()||!"websocket".equals(request.headers().get("Upgrade"))) {
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(web_socket_url, null, false);
        handshaker = factory.newHandshaker(request);
        if(handshaker==null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        }else {
            handshaker.handshake(ctx.channel(), request);
        }
    }
    private void sendHttpResponse(ChannelHandlerContext ctx,FullHttpRequest request,DefaultFullHttpResponse response) {
        if(response.status().code()!=200) {
            ByteBuf buf = Unpooled.copiedBuffer(response.status().toString(),CharsetUtil.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
        }
        ChannelFuture future = ctx.channel().writeAndFlush(response);
        if(response.status().code()!=200) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端与服务端建立连接");
        NettyConfig.group.add(ctx.channel());
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端与服务端断开连接");
        NettyConfig.group.remove(ctx.channel());
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}

WebsocketChildChannelHandler.java

package com.xxx.netty.websocket;
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;
public class WebSocketChildChannelHandler extends ChannelInitializer {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(“http-codec”,new HttpServerCodec());
ch.pipeline().addLast(“aggregator”,new HttpObjectAggregator(65536));
ch.pipeline().addLast(“http-chunked”,new ChunkedWriteHandler());
ch.pipeline().addLast(“handler”,new WebSocketChannelHandler());
}
}

Main.java

package com.xxx.netty.websocket;
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;
public class Main {
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new WebSocketChildChannelHandler());
System.out.println(“server running and waiting for client to connect.”);
Channel channel = bootstrap.bind(8888).sync().channel();
channel.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}

index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta 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://127.0.0.1:8888/websocket");
            socket.onmessage = function(event){
                var ta = document.getElementById("messageContent");
                ta.value +=event.data+"\r\n";
            }
            socket.onopen = function(event){
                var ta = document.getElementById("messageContent");
                ta.value = "您当前的浏览器支持WebSocket,请进行后续操作\r\n";
            }
            socket.onclose =function (event) {
                var ta = document.getElementById("messageContent");
                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"/>
         <input type="button" value="发送消息" onclick="send(this.form.message.value)"/>
         <hr color="red"/>
         <textarea style="width:1024px;height:300px;" id="messageContent"></textarea>
     </form>
</body>
</html>