Java随笔记 - BIO，Socket实例，实现多笔数据的传输

﹏ヽ暗。殇╰゛Y 2022-12-22 08:47 133阅读 0赞

###  Java随笔记 - BIO，Socket实例，实现多笔数据的传输 ###

#### 实现多笔数据的传输 ####

*  在之前的博客（[Java随笔记 - BIO Socket 编程实例][Java_ - BIO Socket]）中，给出了两个简单的Socket编程实例。一个是实现了字节流传输，其中由于服务端并不清楚客户端的数据传输在什么时候结束，所以在客户端代码的第23行可以看到，通过调用socket.shutdownOutput( )告知服务端，客户端的数据已经传输完毕了，相应的在服务端的代码第24行可以看到，服务端通过调用socket.isInputShutdown( )来检测客户端是否完成其数据的传输。
 *  虽然上述方法可以实现数据传输完毕后的标识效果，但是一旦客户端调用了socket.shutdownOutput( )方法后，就不能再使用当前的连接向服务端发送其他数据，如果想要发送新的数据，那只能发起新的连接，而发起连接的代价，可想而知。所以，我们肯定尽可能的在一个连接上进行多笔数据的传输。但是问题也随之而来，传输多个字符串或者多个文件时，我们需要制定一个传输协议，以标识每笔数据的边界。常用的数据边界标识方法有两种：一种是在每笔数据的最后添加一个结束标识；另一种则是在发送每笔数据之前，先发送这笔数据的长度。
 *  在[Java随笔记 - BIO Socket 编程实例][Java_ - BIO Socket]中，另一个编程实例实现了字符流传输，它就是采用了上述中的第一种方法，使用了换行符来作为每笔字符串数据的结束符。但是这一策略在字节流传输中并不适用，因为类似的结束符十分容易与文件中的数据发生冲突，最后导致数据传输错误。所以在字节流传输中，应采用上述中的第二种方法。
 *  下面的代码将会使用这一策略，在连接建立后，客户端向服务端发送多个文件。客户端在发送每一个文件之前，需要计算当前文件的字节数（使用int类型变量进行存储），然后转换成4个字节发送给服务端，以此让服务端得知接下来将传输的文件大小。服务端在连接建立后，首先从输入流中读取4个字节，并将其转换成int类型，就能知道接下来客户端将要传输的文件大小，然后通过循环读取以及累计，将客户端传输过来的数据写入到服务端的文件中，在累计达到文件大小后，则完成此次数据传输，向客户端返回一个"ok"字符串。后续服务端则继续尝试从输入流中读取4个字节的数据，开始新的数据的传输，以此类推。而客户端则在完成所有文件的传输后，调用socket.shutdownOutput( )告知服务端其数据传输完毕，服务端则会通过socket.isInputShutdown( )来决定是否结束数据读取的循环。具体的代码实现如下：
    
        // 客户端代码 Client.java
        public class Client { 
        
            public static final Logger logger = LoggerFactory.getLogger(Client.class);
        
            public static byte[] intToBytearray(int num) { 
                return new byte[] { (byte)((num >> 24) & 0xFF), (byte)((num >> 16) & 0xFF),
                        (byte)((num >> 8) & 0xFF), (byte)(num & 0xFF)};
            }
        
            public static void main(String[] args) { 
                BufferedReader bufferedReader = null;
                BufferedInputStream bufferedInputStream = null;
                BufferedOutputStream bufferedOutputStream = null;
                byte[] buffer = new byte[16];
                Socket socket = null;
                try { 
                    socket = new Socket("127.0.0.1" ,8080);
                    bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    bufferedOutputStream = new BufferedOutputStream(socket.getOutputStream());
                    for (int i = 1; i < 4; i++) { 
                        String fileName = "tmpClient" + i + ".txt";
                        bufferedInputStream = new BufferedInputStream(new FileInputStream(new File(fileName)));
                        int t = bufferedInputStream.available();
                        logger.info("send file {}, size = {}", fileName, t);
                        bufferedOutputStream.write(intToBytearray(t));
                        int r = 0;
                        while ((r = bufferedInputStream.read(buffer)) > 0) { 
                            t -= r;
                            logger.info("read {} bytes from file, {} bytes remain.", r, t);
                            bufferedOutputStream.write(buffer, 0, r);
                            bufferedOutputStream.flush();
                        }
                        String s = bufferedReader.readLine();
                        if ("ok".equalsIgnoreCase(s)) { 
                            logger.info("transport file {} ok", fileName);
                        } else { 
                            logger.error("transport file {} fail", fileName);
                        }
                    }
                    socket.shutdownOutput();
                } catch (Exception e) { 
                    e.printStackTrace();
                } finally { 
                    StreamUtil.close(bufferedReader);
                    StreamUtil.close(bufferedOutputStream);
                    StreamUtil.close(bufferedInputStream);
                    StreamUtil.close(socket);
                }
            }
        
        }
    
        // 服务端代码 Processor.java 实现了Runnable接口
        // Server每接收到一个客户端的请求，就会实例化一个Processor对象，开启一个线程进行处理
        
        public class Processor implements Runnable { 
        
            public static final Logger logger = LoggerFactory.getLogger(Processor.class);
        
            private Socket socket;
        
            public Processor(Socket socket) { 
                this.socket = socket;
            }
        
            public static int byteArrayToInt(byte[] arr) { 
                return (arr[0] & 0xFF) << 24 | (arr[1] & 0xFF) << 16 |
                        (arr[2] & 0xFF) << 8 | (arr[3] & 0xFF);
            }
        
            @Override
            public void run() { 
                PrintWriter printWriter = null;
                BufferedInputStream bufferedInputStream = null;
                BufferedOutputStream bufferedOutputStream = null;
                byte[] buffer = new byte[16];
                int cnt = 0;
                byte[] header = new byte[4];
                try { 
                    bufferedInputStream = new BufferedInputStream(socket.getInputStream());
                    printWriter = new PrintWriter(socket.getOutputStream(), true);
                    while (!socket.isInputShutdown() && bufferedInputStream.read(header) == 4) { 
                        int t = byteArrayToInt(header);
                        if (t < 1) { 
                            printWriter.println("error");
                            break;
                        }
                        logger.info("incoming file size = {}", t);
                        String fileName = "tmpServer" + cnt + ".txt";
                        bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(new File(fileName)));
                        int r = 0;
                        while (t > 0 && (r = bufferedInputStream.read(buffer)) > 0) { 
                            t -= r;
                            logger.info("t bytes remain, {}", t);
                            bufferedOutputStream.write(buffer, 0, r);
                            bufferedOutputStream.flush();
                        }
                        logger.info("received file {}", fileName);
                        printWriter.println("ok");
                        cnt++;
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                } finally { 
                    StreamUtil.close(printWriter);
                    StreamUtil.close(bufferedOutputStream);
                    StreamUtil.close(bufferedInputStream);
                    StreamUtil.close(socket);
                }
            }
        }
    
        // 服务端代码 Server.java
        // 主要就是负责监听来自客户端的连接请求
        
        public class Server implements Runnable { 
        
            public static final Logger logger = LoggerFactory.getLogger(Server.class);
        
            private final ExecutorService threadPool = Executors.newCachedThreadPool();
            private ServerSocket serverSocket = null;
        
            public void start() throws IOException { 
                serverSocket = new ServerSocket(8080);
                threadPool.execute(this);
            }
        
            @Override
            public void run() { 
                Socket socket = null;
                try { 
                    while ((socket = serverSocket.accept()) != null) { 
                        logger.info("client {} connected.", socket.getRemoteSocketAddress());
                        threadPool.execute(new Processor(socket));
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
        
            public void close() { 
                if (serverSocket != null && !serverSocket.isClosed()) { 
                    try { 
                        serverSocket.close();
                    } catch (IOException e) { 
                        logger.error("error on close.", e);
                    }
                }
                threadPool.shutdown();
            }
        
            public static void main(String[] args) { 
                Server server = new Server();
                BufferedReader keyboardReader = null;
                try { 
                    server.start();
        
                    System.out.println("type 'exit' to end.");
        
                    keyboardReader = new BufferedReader(new InputStreamReader(System.in));
                    String cmd = null;
                    while ((cmd = keyboardReader.readLine()) != null) { 
                        if ("exit".equalsIgnoreCase(cmd)) { 
                            break;
                        }
                    }
                } catch (Exception e) { 
                    e.printStackTrace();
                } finally { 
                    StreamUtil.close(keyboardReader);
                    server.close();
                }
            }
        
        }
    
        // 工具类
        	
        public class StreamUtil { 
        	
            public static final Logger logger = LoggerFactory.getLogger(StreamUtil.class);
        	
            public static void close(Closeable stream) { 
                if (stream == null) { 
                    return ;
                }
                try { 
                    stream.close();
                } catch (Exception e) { 
                    logger.error("errors on close {}", stream.getClass().getName(), e);
                }
            }
        	
        }

[Java_ - BIO Socket]: https://blog.csdn.net/J__Max/article/details/109690397