Java IO编程由浅入深 - 1 (bio c/s架构实现)

痛定思痛。 2022-11-06 08:59 53阅读 0赞

### Java实现bio服务端与客户端互相通讯 ###

*  目录
 *  Java服务端代码实现
 *  Java 客户端代码实现
 *  maven依赖如下

# 目录 #

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

# Java服务端代码实现 #

package com.lhstack.bio;
    
    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.charset.StandardCharsets;
    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 boolean initState;
    
        public Server(int port,int nThreads){ 
            this.port = port;
            this.nThreads = nThreads;
            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);
                                InputStream in = client.getInputStream();
                                byte[] bytes = new byte[512];
                                int len = 0;
                                //这里会一直阻塞,直到客户端关闭
                                while((len  = in.read(bytes)) > 0){ 
                                    System.out.println("收到客户端发来的消息: " + new String(bytes,0,len, StandardCharsets.UTF_8));
                                    OutputStream out = client.getOutputStream();
                                    out.write(bytes,0,len);
                                    out.flush();
                                }
                            }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);
            server.start();
        }
    }

# Java 客户端代码实现 #

package com.lhstack.bio;
    
    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.charset.StandardCharsets;
    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;
    
        public Client(String host,int port){ 
            this.port = port;
            this.host = host;
            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");
                            InputStream in = this.socket.getInputStream();
                            int len = 0;
                            byte[] bytes = new byte[512];
                            while((len = in.read(bytes)) > 0){ 
                                System.out.println("收到服务端返回的消息: " + new String(bytes,0,len, StandardCharsets.UTF_8));
                            }
                        }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) { 
            Client client = new Client("localhost",8080);
            Client connect = client.connect();
            Scanner scanner = new Scanner(System.in);
            new Thread(() ->{ 
                while(!Thread.interrupted()){ 
                    System.out.println("请输入你要发送的消息:");
                    String msg = scanner.nextLine();
                    connect.send(msg);
                }
            }).start();
        }
    }

# maven依赖如下 #

<dependency>
       <groupId>ch.qos.logback</groupId>
       <artifactId>logback-classic</artifactId>
       <version>1.2.3</version>
    </dependency>

`分别运行server与client文件里面的main方法，达到以下效果`  
server

22:19:50.602 [main] INFO com.lhstack.bio.Server - bio server start success
    22:19:55.056 [pool-1-thread-1] INFO com.lhstack.bio.Server - client connection success,client Socket[addr=/127.0.0.1,port=56316,localport=8080]

client

22:19:55.056 [Thread-0] INFO com.lhstack.bio.Client - connect server success
    请输入你要发送的消息:

这时我们在客户端输入消息并按回车，得到以下效果  
server

收到客户端发来的消息: hello world

client

请输入你要发送的消息:
    hello world
    请输入你要发送的消息:
    收到服务端返回的消息: hello world

通过以上代码，我们实现了一个简单的c/s架构，客户端到服务端简单通信的程序  
下期对将对此代码进行扩展，实现自定义编解码器和handler，方便以后扩展我们的协议

[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