Java IO编程由浅入深 - 9 (bio实现http升级到websocket )

水深无声 2022-10-22 12:53 133阅读 0赞

### http升级到websocket ###

*  目录
 *  编写websocket handler
 *  编写websocket 编解码器
 *  服务端测试代码
 *  测试浏览器通过http请求，能否收到正确消息
 *  使用websocket客户端连接test地址
 *  使用websocket客户端连接hello地址
 *  连接未配置的地址

# 目录 #

1.  [Java IO编程由浅入深 - 1 (bio c/s架构实现)][Java IO_ - 1 _bio c_s]
2.  [Java IO编程由浅入深 - 2(bio 基于字符串的消息编解码器)][Java IO_ - 2_bio]
3.  [Java IO编程由浅入深 - 3 (bio 基于消息长度的解编码器)][Java IO_ - 3 _bio]
4.  [Java IO编程由浅入深 - 4 (bio http协议解编码器，实现http服务器)][Java IO_ - 4 _bio http_http]
5.  [Java IO编程由浅入深 - 5 (项目架构重构)][Java IO_ - 5]
6.  [Java IO编程由浅入深 - 6 (bio实现C/S聊天室 1 )][Java IO_ - 6 _bio_C_S_ 1]
7.  [Java IO编程由浅入深 - 7 (bio实现C/S聊天室 2 )][Java IO_ - 7 _bio_C_S_ 2]
8.  [Java IO编程由浅入深 - 8 (bio实现C/S聊天室 3 )][Java IO_ - 8 _bio_C_S_ 3]
9.  [Java IO编程由浅入深 - 9 (bio实现http升级到websocket )][Java IO_ - 9 _bio_http_websocket]
10. [Java IO编程由浅入深 - 10 (bio 基于websocket的心跳检测实现 )][Java IO_ - 10 _bio _websocket_]

# 编写websocket handler #

通过websockethandler接收http请求，如果是websocket升级协议，则添加websocket编解码器，同时添加websockethandler  
websocket handler代码如下,构造函数设置handler的方式是因为可能有多个不同业务需要对接websocket，所以为每个业务区分不同handler

package com.lhstack.bio.handler.websocket;
    
    import com.lhstack.bio.channel.ChannelHandlerContext;
    import com.lhstack.bio.channel.SimpleChannelAdapterHandler;
    import com.lhstack.bio.codec.http.HttpRequest;
    import com.lhstack.bio.codec.http.HttpResponse;
    import com.lhstack.bio.codec.websocket.Websocket13MessageCodec;
    import com.lhstack.bio.codec.websocket.frame.AbstractWebSocketFrame;
    
    import java.nio.charset.StandardCharsets;
    import java.security.MessageDigest;
    import java.util.Base64;
    import java.util.Map;
    
    /** * @author lhstack * websocket 处理器 */
    public class WebsocketMessageHandler extends SimpleChannelAdapterHandler<HttpRequest> { 
    
        private final String path;
    
        private final SimpleChannelAdapterHandler<? extends AbstractWebSocketFrame> handler;
    
        /** * websocket协议握手约定的签名key */
        private static final String WEBSOCKET_13_RANDOM_STR = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
    
        public WebsocketMessageHandler(String path, SimpleChannelAdapterHandler<? extends AbstractWebSocketFrame> handler){ 
            this.path = path;
            this.handler = handler;
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception { 
            //websocket握手协议携带的请求头
            if(msg.headers().containsKey("Upgrade") && msg.headers().containsKey("Connection")){ 
                String uri = msg.uri();
                int index = uri.indexOf("?");
                if(index > 0){ 
                    uri = uri.substring(0,index);
                }
                if(uri.equals(this.path)){ 
                    upgrade(msg,ctx);
                    return ;
                }
            }
            ctx.fireChannelRead(msg);
        }
    
        private void upgrade(HttpRequest msg, ChannelHandlerContext ctx) throws Exception{ 
            HttpResponse res = new HttpResponse("HTTP/1.1",101);
            Map<String, String> headers = res.headers();
            headers.put("Upgrade","websocket");
            headers.put("Connection","Upgrade");
            String key = msg.headers().get("Sec-WebSocket-Key");
            key += WEBSOCKET_13_RANDOM_STR;
            byte[] sha1 = MessageDigest.getInstance("SHA1").digest(key.getBytes(StandardCharsets.UTF_8));
            headers.put("Sec-WebSocket-Accept", Base64.getEncoder().encodeToString(sha1));
            //发送握手响应
            ctx.writeAndFlush(res);
            //然后添加设置的handler和websocket解编码器
            ctx.pipeline().addFirst(this.handler)
                    .addFirst(new Websocket13MessageCodec());
        }
    }

# 编写websocket 编解码器 #

用来解析websocket传输协议，区分成不同handler

package com.lhstack.bio.codec.websocket;
    
    import com.lhstack.bio.channel.ChannelHandlerContext;
    import com.lhstack.bio.codec.ByteToMessageCodec;
    import com.lhstack.bio.codec.websocket.frame.*;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.util.ReferenceCountUtil;
    
    import java.util.List;
    
    /** * @author lhstack * websocket解编码器，协议请参考RFC4644 */
    public class Websocket13MessageCodec extends ByteToMessageCodec<AbstractWebSocketFrame> { 
    
        enum State { 
            READING_FIRST,
            READING_SECOND,
            READING_SIZE,
            MASKING_KEY,
            PAYLOAD
        }
    
        private static final byte OPCODE_CONT = 0x0;
        private static final byte OPCODE_TEXT = 0x1;
        private static final byte OPCODE_BINARY = 0x2;
        private static final byte OPCODE_CLOSE = 0x8;
        private static final byte OPCODE_PING = 0x9;
        private static final byte OPCODE_PONG = 0xA;
    
        private boolean frameFinalFlag;
    
        private boolean frameMasked;
    
        private int frameRsv;
    
        private int frameOpcode;
    
        private long framePayloadLength;
    
        private byte[] maskingKey;
    
        private int framePayloadLen1;
    
        private State state = State.READING_FIRST;
    
        @Override
        protected void messageEncoder(ChannelHandlerContext ctx, AbstractWebSocketFrame msg, ByteBuf out) throws Exception { 
            int opcode = msg.getOpcode();
            int b0 = 0;
            //设置fin
            b0 |= msg.fin() << 7;
            //设置rsv
            b0 |= msg.rsv() % 8 << 4;
            //设置opcode
            b0 |= opcode % 128;
            out.writeByte(b0);
            int len = msg.content().readableBytes();
            //设置长度，如果进行了掩码，则添加掩码
            if(len <= 125){ 
                out.writeByte(msg.masked() ? 0x80 | len : len);
            }else if(len <= 0xFFFF){ 
                //如果小于32767,则是一个short长度，同样也要设置掩码
                out.writeByte(msg.masked() ? 0xFE : 126);
                out.writeByte(len >>> 8 & 0xFF);
                out.writeByte(len & 0xFF);
            }else{ 
                //大于127，则是一个long长度的内容
                out.writeByte(msg.masked() ? 0xFF : 127)
                        .writeLong(len);
            }
            //如果设置了掩码，则需要对内容进行编码
            if(msg.masked()){ 
                ByteBuf content = msg.content();
                int length = content.readableBytes();
                ByteBuf maskArray = Unpooled.buffer(4).writeInt((int) (Math.random() * Integer.MAX_VALUE));
                for (int j = 0; j < length; j++) { 
                    content.setByte(j,content.getByte(j) ^ maskArray.getByte(j % 4));
                }
                ReferenceCountUtil.safeRelease(maskArray);
            }
            out.writeBytes(msg.content());
        }
    
        @Override
        protected void messageDecoder(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) { 
            if(this.state == State.READING_FIRST){ 
                if(!buf.isReadable()){ 
                    return ;
                }
                byte b = buf.readByte();
                this.framePayloadLength = 0;
                //fin 取第一位,判断fin是否为0
                this.frameFinalFlag = (b & 0x80) != 0;
                //rsv 取3位，向右移动四位，则是第二位，第三位，第四位 从第二位开始到第四位，然后向右移动4位，则为低三位
                this.frameRsv = (b & 0x70) >> 4;
                //取最后四位
                this.frameOpcode = b & 0x0F;
                //如果是close，则直接关闭
                if(this.frameOpcode == OPCODE_CLOSE){ 
                    ctx.channel().close();
                    return ;
                }
                this.state = State.READING_SECOND;
            }
            if(this.state == State.READING_SECOND){ 
                if(!buf.isReadable()){ 
                    return ;
                }
                byte b = buf.readByte();
                //是否有掩码
                this.frameMasked = (b & 0x80) != 0;
                //读取到的第一次的长度
                this.framePayloadLen1 = b & 0x7F;
                this.state = State.READING_SIZE;
            }
            if(this.state == State.READING_SIZE){ 
                //如果长度等于126，则后面两个字节是长度
                if (this.framePayloadLen1 == 126) { 
                    if (buf.readableBytes() < 2) { 
                        return;
                    }
                    this.framePayloadLength = buf.readUnsignedShort();
                    //如果等于127，则后面八个字节是长度
                } else if (this.framePayloadLen1 == 127) { 
                    if (buf.readableBytes() < 8) { 
                        return;
                    }
                    this.framePayloadLength = buf.readLong();
                } else { 
                    //否则本身就是长度
                    this.framePayloadLength = this.framePayloadLen1;
                }
                this.state = State.MASKING_KEY;
            }
            if(this.state == State.MASKING_KEY){ 
                //如果有掩码
                if (this.frameMasked) { 
                    if (buf.readableBytes() < 4) { 
                        return;
                    }
                    if (this.maskingKey == null) { 
                        this.maskingKey = new byte[4];
                    }
                    buf.readBytes(this.maskingKey);
                }
                this.state = State.PAYLOAD;
            }
            if(this.state == State.PAYLOAD){ 
                if (buf.readableBytes() < this.framePayloadLength) { 
                    return;
                }
                ByteBuf payloadBuffer = Unpooled.buffer((int) this.framePayloadLength);
                buf.readBytes(payloadBuffer);
                if(this.frameMasked){ 
                    int length = payloadBuffer.readableBytes();
                    for (int i = 0; i < length; i++) { 
                        payloadBuffer.setByte(i,payloadBuffer.getByte(i) ^ this.maskingKey[i % 4]);
                    }
                }
                switch (this.frameOpcode){ 
                    case OPCODE_TEXT:
                        out.add(new TextWebSocketFrame(payloadBuffer));break;
                    case OPCODE_BINARY:
                        out.add(new BinaryWebSocketFrame(payloadBuffer));break;
                    case OPCODE_PING:
                        out.add(new PingWebSocketFrame());break;
                    case OPCODE_PONG:
                        out.add(new PongWebSocketFrame());break;
                    default:{ 
    
                    }
                }
                ReferenceCountUtil.safeRelease(payloadBuffer);
                payloadBuffer = null;
                this.state = State.READING_FIRST;
            }
        }
    }

# 服务端测试代码 #

public static void websocketServer() throws Exception { 
            ServerBootStrap serverBootStrap = new ServerBootStrap(8080);
            serverBootStrap.group(Executors.newFixedThreadPool(200),new DefaultEventLoopGroup())
                    .handler(new ChannelInitializeHandler() { 
                        @Override
                        public void initializer(Channel ch) throws Exception { 
                            ch.pipeline()
                                    .addLast(new HttpServerMessageCodec())
                                    .addLast(new WebsocketMessageHandler("/test", new SimpleChannelAdapterHandler<TextWebSocketFrame>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
                                            System.out.println(msg.text());
                                            ctx.writeAndFlush(new TextWebSocketFrame("this is test message"));
                                        }
    
                                        @Override
                                        public void channelRemove(ChannelHandlerContext ctx) throws Exception { 
                                            System.out.println(ctx);
                                        }
    
                                        @Override
                                        public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                            System.out.println(ctx);
                                            throwable.printStackTrace();
                                        }
                                    }))
                                    .addLast(new WebsocketMessageHandler("/hello", new SimpleChannelAdapterHandler<TextWebSocketFrame>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
                                            System.out.println(msg.text());
                                            ctx.writeAndFlush(new TextWebSocketFrame("this is hello world message"));
                                        }
    
                                        @Override
                                        public void channelRemove(ChannelHandlerContext ctx) throws Exception { 
                                            System.out.println(ctx);
                                        }
    
                                        @Override
                                        public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                            System.out.println(ctx);
                                            throwable.printStackTrace();
                                        }
                                    }))
                                    .addLast(new SimpleChannelAdapterHandler<HttpRequest>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception { 
                                            System.out.println(msg);
                                            ctx.writeAndFlush(new HttpResponse("HTTP/1.1",200,Unpooled.wrappedBuffer("hello world".getBytes(StandardCharsets.UTF_8))));
                                        }
                                    });
                        }
                    }).start();
        }

# 测试浏览器通过http请求，能否收到正确消息 #

可以看到，http请求能处理的，现在使用websocket客户端连接试试  
![在这里插入图片描述][20210317130844756.png]  
![在这里插入图片描述][2021031713085491.png]

# 使用websocket客户端连接test地址 #

可以看到，test地址能通，现在试试hello地址  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70]

# 使用websocket客户端连接hello地址 #

可以看到，hello地址也能正确的到正确的handler，现在连一个未配置的地址试试  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 1]

# 连接未配置的地址 #

可以看到，未配置的地址是无法进行通讯的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 2]  
下一期我将实现基于websocket群聊案例

[Java IO_ - 1 _bio c_s]: https://blog.csdn.net/qq_42413011/article/details/114553235
[Java IO_ - 2_bio]: https://blog.csdn.net/qq_42413011/article/details/114579502
[Java IO_ - 3 _bio]: https://blog.csdn.net/qq_42413011/article/details/114636827
[Java IO_ - 4 _bio http_http]: https://blog.csdn.net/qq_42413011/article/details/114646159
[Java IO_ - 5]: https://blog.csdn.net/qq_42413011/article/details/114802496
[Java IO_ - 6 _bio_C_S_ 1]: https://blog.csdn.net/qq_42413011/article/details/114852932
[Java IO_ - 7 _bio_C_S_ 2]: https://blog.csdn.net/qq_42413011/article/details/114878895
[Java IO_ - 8 _bio_C_S_ 3]: https://blog.csdn.net/qq_42413011/article/details/114896029
[Java IO_ - 9 _bio_http_websocket]: https://blog.csdn.net/qq_42413011/article/details/114925184
[Java IO_ - 10 _bio _websocket_]: https://blog.csdn.net/qq_42413011/article/details/114947311
[20210317130844756.png]: /images/20221022/8b68b77f2e6945e28cab4ada3069872d.png
[2021031713085491.png]: /images/20221022/17abf1fbd5124ae28907a4db718c517a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70]: /images/20221022/3d85aaa9f9044f66924e53a3fe1fe0fa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 1]: /images/20221022/c1124a7e02a04932a8629ddc20477a97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 2]: /images/20221022/2b639a5ee86a4b629f30268fd8b844cd.png