Netty实现websocket示例

谁借莪１个温暖的怀抱￠ 2022-03-12 10:08 516阅读 0赞

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

**工程结构：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70][]

**pom.xml**

<dependency>  
        <groupId>io.netty</groupId>  
        <artifactId>netty-all</artifactId>  
        <version>4.1.27.Final</version>  
</dependency>

**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<SocketChannel> \{  
    @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>

**前端页面:**

![20190305183222731.gif][]

**服务端打印信息： **

![2019030518303961.png][]

这里使用的是netty-all-4.1.27.Final版本，其中handler需要实现的方法是channelRead0(ctx,msg);如果使用的是netty-all-5.0.0.Alpha1版本，方法变为messageReceived(ctx,msg)。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70]: /images/20220312/788e54bc2a32431a8df7db258848f2bd.png
[20190305183222731.gif]: /images/20220312/d7054f4b75414908a1408d0b5ca02f77.png
[2019030518303961.png]: /images/20220312/d982cc5298324acf816eadf855e5e672.png