Java IO编程由浅入深 - 3 (bio 基于消息长度的解编码器)

川长思鸟来 2022-11-07 00:56 10阅读 0赞

### 解编码器,基于长度 ###

*  目录

# 目录 #

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_]

`经过上次的解编码器，我发现一个问题，当多个连接进来之后，会共享解编码器和handler，会出现多线程并发问题，因此，这一期我们将之前的server端进行修改，通过函数方式构造编解码器和handler，实现每个连接只会有与连接对应的解编码器和handler，数据由所在连接共享,同时，由于jdk自带的ByteBuffer不能实现自动扩容，因此我们使用netty-buffer工具中的ByteBuf作为数据传输`  
1.编写ObjectProvider接口，定义对象构造函数式接口

package com.lhstack.bio.object;
    
    /** * @author lhstack * @description 对象构造器 * @param <T> */
    public interface ObjectProvider<T> { 
    
        /** * 构造对象 * @return */
        T newBuilder();
    }

2.修改原server.java的代码,使用ObjectProvider构造编解码器和handler，同时替换byteBuffer为ByteBuf

package com.lhstack.bio;
    
    import com.lhstack.bio.codec.MessageCodec;
    import com.lhstack.bio.codec.StringMessageCodec;
    import com.lhstack.bio.handler.MessageHandler;
    import com.lhstack.bio.object.ObjectProvider;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.ByteBufUtil;
    import io.netty.buffer.Unpooled;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.net.*;
    import java.nio.ByteBuffer;
    import java.time.LocalDateTime;
    import java.time.format.DateTimeFormatter;
    import java.util.Objects;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /** * @author lhstack * @description bio server * @date 2021/3/8 21:03 */
    public class Server { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(Server.class);
        /** * 监听端口 */
        private final int port;
    
        /** * 运行状态 服务端是否运行 */
        private AtomicBoolean running;
    
        /** * 能接受客户端的连接数 */
        private final int nThreads;
    
        /** * 用线程池来存储客户端，一次最大只能有200个连接进行并发连接 */
        private ExecutorService executorService;
    
        private ServerSocket serverSocket;
    
        /** * 自定义消息编解码器 */
        private final ObjectProvider<MessageCodec> messageCodecProvider;
    
        /** * 使用对象构造器，保证每个连接会单独存在一个编解码器和handler，防止多线程单例问题 */
        private final ObjectProvider<MessageHandler> messageProvider;
    
        /** * 初始化状态 */
        private boolean initState;
    
        public Server(int port, int nThreads, ObjectProvider<MessageCodec> messageCodecProvider, ObjectProvider<MessageHandler> messageProvider){ 
            this.port = port;
            this.nThreads = nThreads;
            this.messageCodecProvider = messageCodecProvider;
            this.messageProvider = messageProvider;
            this.init();
        }
    
        /** * 初始化方法 * TCP_NODELAY 设置客户端无延迟 * SO_RCVBUF 设置服务端一次最大能接收的字节数 * SO_SNDBUF 设置服务端一次最大能发送的字节数 */
        private void init() { 
            try { 
                this.running = new AtomicBoolean(false);
                this.initState = true;
                this.executorService = Executors.newFixedThreadPool(this.nThreads);
                this.serverSocket = new ServerSocket();
                this.serverSocket
                        .setOption(StandardSocketOptions.SO_RCVBUF,512);
            } catch (IOException e) { 
                initState = false;
                LOGGER.error("bio server init failed,throw error => {}",e.getMessage(),e);
                throw new RuntimeException(e);
            }
        }
    
        /** * 关闭bio服务 */
        public void stop() throws Exception { 
            if(this.running.get()){ 
                this.serverSocket.close();
                this.running.set(false);
                LOGGER.info("bio server stop success");
            }
        }
    
        /** * 启动bio服务 */
        public void start() throws Exception{ 
            if(!this.initState){ 
                LOGGER.error("bio server init failed, can not start");
                return ;
            }
            if(!this.running.get()){ 
                this.serverSocket.bind(new InetSocketAddress(this.port));
                //绑定异常，就不会设置运行状态
                this.running.set(true);
                LOGGER.info("bio server start success");
                try{ 
                    while(this.running.get()){ 
                        Socket client = this.serverSocket.accept();
                        //连接进来，构造messageHandler
                        MessageHandler<Object> messageHandler = this.messageProvider.newBuilder();
                        MessageCodec<Object> messageCodec = this.messageCodecProvider.newBuilder();
                        executorService.submit(() ->{ 
                            try{ 
                                messageHandler.channelActive(client);
                                //设置客户端无延迟
                                client.setOption(StandardSocketOptions.TCP_NODELAY,true);
                                //设置客户端一次发送的最大字节数
                                client.setOption(StandardSocketOptions.SO_SNDBUF,512);
                                LOGGER.info("client connection success,client {}",client);
                                OutputStream out = client.getOutputStream();
                                InputStream in = client.getInputStream();
                                byte[] bytes = new byte[512];
                                int len = 0;
                                ByteBuf buffer = Unpooled.buffer(512);
                                //这里会一直阻塞,直到客户端关闭
                                while((len  = in.read(bytes)) > 0){ 
                                    buffer.writeBytes(bytes,0,len);
                                    Object msg = messageCodec.messageDecoder(buffer);
                                    //解码成功，并有返回值才处理
                                    if(Objects.nonNull(msg)){ 
                                        //如果不想输出消息到客户端，直接返回null即可
                                        Object rst = messageHandler.channelRead(msg,client);
                                        if(Objects.nonNull(rst)){ 
                                            ByteBuf buf = messageCodec.messageEncoder(rst);
                                            byte[] result = new byte[buf.readableBytes()];
                                            buf.readBytes(result);
                                            out.write(result);
                                            out.flush();
                                        }
                                        //获取剩下的内容，构建新的byteBuf给解码器使用
                                        byte[] bs = ByteBufUtil.getBytes(buffer);
                                        //回收之前老的buf
                                        buffer.release();
                                        buffer = Unpooled.buffer(512);
                                        buffer.writeBytes(bs);
                                    }
                                }
                            }catch (Exception e){ 
                                if(Objects.nonNull(client)){ 
                                    try{ 
                                        messageHandler.channelClose(client);
                                        client.close();
                                    }catch (Exception ex){ 
                                        LOGGER.error("close client throw error {}",ex.getMessage(),ex);
                                    }
                                }
                                LOGGER.error("client an exception is thrown during processing,close client {},throw error {} ",client,e.getMessage(),e);
                            }
                        });
                    }
                }catch (Exception e){ 
                    //启动失败,初始化运行状态
                    this.running.set(false);
                    //这里因为调用stop方法的时候，会抛出SocketException，所以为了友好提示，去掉这个异常
                    if(!(e instanceof SocketException)){ 
                        LOGGER.error("bio server accept throw error {}",e.getMessage(),e);
                    }
                }
            }else{ 
                LOGGER.warn("bio server running state is true");
            }
        }
    }

3.修改对应编解码器里面的ByteBuffer为ByteBuf

package com.lhstack.bio.codec;
    
    import io.netty.buffer.ByteBuf;
    
    /** * 编码器 * @author lhstack */
    public interface MessageCodec<T> extends Cloneable { 
    
        /** * 将msg编码成ByteBuffer * @param msg * @return ByteBuf */
        ByteBuf messageEncoder(T msg);
    
        /** * messageDecoder * 将ByteBuffer解码成Msg * @param buf * @return <T> */
        Object messageDecoder(ByteBuf buf);
    
    }

4.编写基于长度的解编码的LengthMessageCodec.java

package com.lhstack.bio.codec;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    
    /** * 消息长度的codec * @author lhstack */
    public class LengthMessageCodec implements MessageCodec<ByteBuf> { 
    
        @Override
        public ByteBuf messageEncoder(ByteBuf msg) { 
            int length = msg.readableBytes();
            //申请容量
            ByteBuf buf = Unpooled.buffer(length + 4);
            //写入消息长度和消息体
            buf.writeInt(length)
                    .writeBytes(msg);
            return buf;
        }
    
        @Override
        public Object messageDecoder(ByteBuf buf) { 
            //标记读取的位置
            buf.markReaderIndex();
            //判断是否有一个int字节长度的内容
            if(buf.isReadable() && buf.readableBytes() >= 4){ 
                //读取消息长度
                int length = buf.readInt();
                //如果buf里面拥有读取长度的内容，则读出内容
                if(buf.readableBytes() >= length){ 
                    ByteBuf result = Unpooled.buffer(length);
                    buf.readBytes(result);
                    return result;
                }//重置读取索引
                buf.resetReaderIndex();
            }
            return null;
        }
    }

5.我们创建一个server端和服务端分别测试lengthMessageCodec的效果

server端

public static void main(String[] args) throws Exception { 
            Server server = new Server(8080, 200, LengthMessageCodec::new, () -> new MessageHandler<ByteBuf>() { 
                @Override
                public Object channelRead(ByteBuf msg, Socket socket) throws Exception { 
                    byte[] bytes = new byte[msg.readableBytes()];
                    msg.readBytes(bytes);
                    System.out.println(new String(bytes));
                    return Unpooled.wrappedBuffer(("我是服务端，我收到你的消息了，当前时间是: " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))).getBytes(StandardCharsets.UTF_8));
                }
            });
            server.start();
        }

client端

public static void main(String[] args) { 
            Scanner scanner = new Scanner(System.in);
            Client client = new Client("localhost", 8080, new LengthMessageCodec(), (MessageHandler<ByteBuf>) (msg, socket) -> { 
                System.out.println("收到服务端的消息是: " + new String(ByteBufUtil.getBytes(msg)));
                System.out.println("请输入你要发送的消息:");
                //如果不想输出消息到服务端，返回null即可
                return Unpooled.wrappedBuffer(scanner.nextLine().getBytes(StandardCharsets.UTF_8));
            });
            Client connect = client.connect();
            //上面的handler是需要收到消息才会触发，所以需要发送一条消息触发上面的handler
            connect.send(Unpooled.wrappedBuffer("hello world".getBytes(StandardCharsets.UTF_8)));
        }

测试当消息体大于512的时候，会不会解码成功，获取完整的消息  
client 控制台输入以下内容，看服务端是否收到以下完整的内容  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70]  
可以看到，以下server端是完整的获取到客户端发送的消息，可以说明此次我们的沾包问题是完整的解决了的  
![在这里插入图片描述][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]  
下一期，我将实现一个简单的http解编码器，通过浏览器想服务端发送请求，并响应数据

[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
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70]: /images/20221023/36bd3796af1945a2a48b8f99d6641f3d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 1]: /images/20221023/5bfb3450ad84407db4b3e62885ebe5fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDEzMDEx_size_16_color_FFFFFF_t_70 2]: /images/20221023/76e60a24cd63434e84a567fa46e0de03.png