Java IO编程由浅入深 - 2(bio 基于字符串的消息编解码器)

你的名字 2022-11-06 11:54 259阅读 0赞

### 这一期我们实现自定义编解码器和业务处理器 ###

*  目录
 *  创建编解码器的接口
 *  实现简单的StringMessageCodec
 *  定义我们的handler接口
 *  修改我们server.java的代码
 *  客户端同服务端类似，也加上相应代码

# 目录 #

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

# 创建编解码器的接口 #

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

# 实现简单的StringMessageCodec #

package com.lhstack.bio.codec;
    
    import java.nio.ByteBuffer;
    import java.nio.charset.StandardCharsets;
    
    /** * @author lhstack * @description StringMessageCodec，对字符串的数据进行编码 */
    public class StringMessageCodec implements MessageCodec<String>{ 
    
        @Override
        public ByteBuffer messageEncoder(String msg) { 
            byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
            ByteBuffer byteBuffer = ByteBuffer.allocateDirect(bytes.length);
            byteBuffer.put(bytes);
            byteBuffer.flip();
            return byteBuffer;
        }
    
        @Override
        public String messageDecoder(ByteBuffer buf) { 
            byte[] bytes = new byte[buf.remaining()];
            buf.get(bytes);
            return new String(bytes,StandardCharsets.UTF_8);
        }
    }

# 定义我们的handler接口 #

package com.lhstack.bio.handler;
    
    /** * 处理器，接受解码后的消息，返回解码后的消息 * @author lhstack * @param <T> */
    public interface MessageHandler <T>{ 
    
        /** * 处理消息 * @param msg * @return T * @throws Exception */
        Object handler(T msg) throws Exception;
    }

# 修改我们server.java的代码 #

在start和构造方法中方便加入MessageCodec和MessageHandler

package com.lhstack.bio;
    
    import com.lhstack.bio.codec.MessageCodec;
    import com.lhstack.bio.codec.StringMessageCodec;
    import com.lhstack.bio.handler.MessageHandler;
    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 MessageCodec<Object> messageCodec;
    
        private final MessageHandler<Object> messageHandler;
    
        /** * 初始化状态 */
        private boolean initState;
    
        public Server(int port, int nThreads, MessageCodec<?> messageCodec, MessageHandler<?> messageHandler){ 
            this.port = port;
            this.nThreads = nThreads;
            this.messageCodec = (MessageCodec<Object>) messageCodec;
            this.messageHandler = (MessageHandler<Object>) messageHandler;
            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();
                        executorService.submit(() ->{ 
                            try{ 
                                //设置客户端无延迟
                                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;
                                ByteBuffer buffer = ByteBuffer.allocate(512);
                                //这里会一直阻塞,直到客户端关闭
                                while((len  = in.read(bytes)) > 0){ 
                                    buffer.put(bytes,0,len);
                                    buffer.flip();
                                    Object msg = this.messageCodec.messageDecoder(buffer);
                                    //解码成功，并有返回值才处理
                                    if(Objects.nonNull(msg)){ 
                                        //如果不想输出消息到客户端，直接返回null即可
                                        Object rst = this.messageHandler.handler(msg);
                                        if(Objects.nonNull(rst)){ 
                                            ByteBuffer buf = this.messageCodec.messageEncoder(rst);
                                            byte[] result = new byte[buf.remaining()];
                                            buf.get(result);
                                            out.write(result);
                                            out.flush();
                                        }
                                        buffer.clear();
                                    }
                                }
                            }catch (Exception e){ 
                                if(Objects.nonNull(client)){ 
                                    try{ 
                                        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");
            }
        }
    
        public static void main(String[] args) throws Exception { 
            Server server = new Server(8080, 200, new StringMessageCodec(), (MessageHandler<String>) msg -> { 
                System.out.println("收到客户端的消息： " + msg);
                return "我是服务端，我收到你的消息了，当前时间是: " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            });
            server.start();
        }
    }

# 客户端同服务端类似，也加上相应代码 #

package com.lhstack.bio;
    
    import com.lhstack.bio.codec.MessageCodec;
    import com.lhstack.bio.codec.StringMessageCodec;
    import com.lhstack.bio.handler.MessageHandler;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.net.InetSocketAddress;
    import java.net.Socket;
    import java.net.StandardSocketOptions;
    import java.nio.ByteBuffer;
    import java.nio.charset.StandardCharsets;
    import java.util.AbstractList;
    import java.util.Objects;
    import java.util.Scanner;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /** * @author lhstack * @description bio client * @date 2021/3/8 21:10 */
    public class Client { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(Client.class);
        /** * 连接端口 */
        private final int port;
    
        /** * 连接地址 */
        private final String host;
    
        /** * 运行状态 客户端是否运行 */
        private AtomicBoolean running;
    
        /** * 初始化状态 */
        private boolean initState;
    
        /** * 向服务端输出消息用的 */
        private OutputStream out;
    
        /** * 客户端对象 */
        private Socket socket;
    
        private Thread clientThread;
    
        private final MessageCodec<Object> messageCodec;
    
        private final MessageHandler<Object> messageHandler;
    
        public Client(String host,int port,MessageCodec<?> messageCodec,MessageHandler<?> messageHandler){ 
            this.port = port;
            this.host = host;
            this.messageCodec = (MessageCodec<Object>) messageCodec;
            this.messageHandler = (MessageHandler<Object>) messageHandler;
            this.init();
        }
    
        private void init() { 
            try { 
                this.running = new AtomicBoolean(false);
                this.initState = true;
                this.socket = new Socket();
                this.socket.setOption(StandardSocketOptions.SO_KEEPALIVE,true);
            } catch (IOException e) { 
                initState = false;
                LOGGER.error("bio server init failed,throw error => {}",e.getMessage(),e);
                throw new RuntimeException(e);
            }
        }
    
        public void stop(){ 
            if(this.running.get()){ 
                try { 
                    this.socket.close();
                    this.clientThread.interrupt();
                    this.running.set(false);
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        public Client connect() { 
            if(!this.initState){ 
                LOGGER.error("bio client init failed, can not connect");
                return null;
            }
            if(!this.running.get()){ 
                try{ 
                    //使用countDownLatch实现同步连接
                    CountDownLatch count = new CountDownLatch(1);
                    this.clientThread = new Thread(() ->{ 
                        try{ 
                            this.socket.connect(new InetSocketAddress(this.host,this.port));
                            this.running.set(true);
                            this.out = this.socket.getOutputStream();
                            count.countDown();
                            LOGGER.info("connect server success");
                            ByteBuffer buf = ByteBuffer.allocate(512);
                            InputStream in = this.socket.getInputStream();
                            OutputStream out = this.socket.getOutputStream();
                            int len = 0;
                            byte[] bytes = new byte[512];
                            while((len = in.read(bytes)) > 0){ 
                                buf.put(bytes,0,len);
                                buf.flip();
                                Object msg = this.messageCodec.messageDecoder(buf);
                                if(Objects.nonNull(msg)){ 
                                    //如果不想输出消息到服务端，直接返回null即可
                                    Object result = this.messageHandler.handler(msg);
                                    if(Objects.nonNull(result)){ 
                                        ByteBuffer buff = this.messageCodec.messageEncoder(result);
                                        byte[] rsBytes = new byte[buff.remaining()];
                                        buff.get(rsBytes);
                                        out.write(rsBytes);
                                        out.flush();
                                    }
                                    buf.clear();
                                }
                            }
                        }catch (Exception e){ 
                            //连接失败，初始化运行状态
                            this.running.set(false);
                            count.countDown();
                            LOGGER.error("client connect throw error {}",e.getMessage(),e);
    
                        }
                    });
                    this.clientThread.start();
                    count.await();
                }catch (Exception e){ 
                    //关闭客户端连接
                    this.stop();
                    LOGGER.error("client connect falied , throw error {}",e.getMessage(),e);
                }
            }else{ 
                LOGGER.warn("bio client running state is true");
            }
            return this;
        }
    
        /** * 发送消息 * @param msg * @return */
        public Client send(String msg){ 
            if(this.running.get()){ 
                try { 
                    this.out.write(msg.getBytes(StandardCharsets.UTF_8));
                    this.out.flush();
                } catch (IOException e) { 
                    LOGGER.error("client send msg throw error {}",e.getMessage(),e);
                }
            }else{ 
                LOGGER.warn("client running is false,cannot send msg");
            }
            return this;
        }
    
        public static void main(String[] args) { 
            Scanner scanner = new Scanner(System.in);
            Client client = new Client("localhost", 8080, new StringMessageCodec(), (MessageHandler<String>) msg -> { 
                System.out.println("收到服务端的消息是: " + msg);
                System.out.println("请输入你要发送的消息:");
                //如果不想输出消息到服务端，返回null即可
                return scanner.nextLine();
            });
            Client connect = client.connect();
            //上面的handler是需要收到消息才会触发，所以需要发送一条消息触发上面的handler
            connect.send("connect");
        }
    }

通过自定义消息解编码器和业务处理handler，简化编码过程，同时让开发人员只需要关注业务相关的解编码器和处理器即可，不需要关心跟io相关的东西  
下一期我将使用MessageCodec实现自定义的解编码器协议，解决沾包问题

[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