Java Socket IO Thread 整合大文件与上传下载

package com.dounine.bigfile; import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.InputStream; import java.io.OutputStream; import java.net.Socket; @SuppressWarnings("all") public class Client { private Socket socket = null; private InputStream is = null; private OutputStream os = null; private DataInputStream dis = null; private DataOutputStream dos = null; private String styleType = null;// 与服务器交互的方式 private String filePath = null;// 文件的存放位置 例如:d:/lake/ public final String READ = "read"; public final String WRITE = "write"; public final String DOWNLOAD = "download"; public final String UPLOAD = "upload"; private Common common = null; public Client(Socket socket, String styleType, String filePath) { common = new Common(); this.styleType = styleType; this.filePath = filePath; if (styleType == null) { System.out.println(" 对服务的操作不能为空!"); System.exit(0); } this.socket = socket; System.out.println(" 与服务建立连接"); try { is = socket.getInputStream(); os = socket.getOutputStream(); dis = new DataInputStream(is); dos = new DataOutputStream(os); } catch (Exception e) { System.out.println(" 流获取异常"); } } public void run() { if (styleType.equals(READ)) { download(); } else if (styleType.equals(WRITE)) { upload(); } } public void upload() {// 上传文件 // 监听对方请求 try { dos.writeUTF("upload"); FileInputStream _fis = null; File file = new File(filePath); File[] files = file.listFiles(); if(files.length==0){ dos.writeUTF("null"); }else{ dos.writeUTF("have"); } for (File file2 : files) { if (file2.length() >= 0 && file2.length() <= 1000) { dos.writeUTF("yes"); uploadSmall(file2); } else { dos.writeUTF("no"); String isBegin = dis.readUTF(); if (isBegin.equals("begin")) { dos.writeUTF("yes"); System.out .println("\n发送文件:" + file2.getName() + " ..."); dos.writeUTF(file2.getName());// 发送文件名 // 开始发送文件 byte[] fileBuf = common.getFileBuffer(); int fileBufferLen = 0; _fis = new FileInputStream(file2.getAbsoluteFile()); long time = System.currentTimeMillis(); while (true) { fileBufferLen = _fis.read(fileBuf); if (fileBufferLen <= 0) { os.write(common.ZeroLength); os.flush(); break; } else { os.write(common .MakeHexIntegerFlag(fileBufferLen)); os.write(fileBuf, 0, fileBufferLen); os.flush(); } dis.readUTF();//断点连接，以免断开 } System.out.println("发送文件:" + file2.getName() + " 完成"); System.out.println("用时：" + (System.currentTimeMillis() - time) + " 毫秒"); } } } if(_fis!=null) _fis.close(); dos.writeUTF(""); } catch (Exception e) { System.out.println("异常退出"); close(); } finally { System.out.println("\n 客户端:" + (socket.getInetAddress().toString()).substring(1) + " 结束连接"); close(); } } public void download() {// 接收文件 try { dos.writeUTF(DOWNLOAD); String filesIsNull = dis.readUTF(); if(filesIsNull.equals("null")){ System.out.println("\n服务器没有可下载文件!"); close(); return; } byte[] flag = common.getFlagBuffer(); while (true) { String is_small = dis.readUTF(); if (is_small.equals("yes")) {// 如果是20-100个字节的文件是小文件 downloadSmall();// 调用接收小文件方法 } else { dos.writeUTF("begin"); String can = dis.readUTF(); if (can.equals("")) { break; } long time = System.currentTimeMillis(); // ---------对方接受了请求------------------- if (can.equals("yes")) { byte[] fileBuf = new byte[32 * 1024 * 1024];// 准备写入文件的缓冲区 byte[] recvBuf = new byte[8 * 1024 * 1024];// 接受网络数据的缓存区 int recvBufferLength = 0;// 已接收的字节数 int destBufferLength = 0;// 期待传输的总字节数 int singleRecvLength = 0;// 单次接收的文件长度 String fileName = null; fileName = dis.readUTF();// 读取文件名 System.out.println("\n接收文件:" + fileName + " ...");// 打印文件名 FileOutputStream file = new FileOutputStream(filePath + fileName); while (true) { is.read(flag); destBufferLength = Integer.parseInt( new String(flag), 16); if (destBufferLength > 0) { recvBufferLength = 0; singleRecvLength = is.read(fileBuf);// 第一次数据直接放到文件缓冲区 // ----循环读取数据直到长度达到预计量--------- while ((recvBufferLength += singleRecvLength) < destBufferLength) { singleRecvLength = is.read(recvBuf); // 放入文件缓存 for (int j = 0; j < singleRecvLength; j++) { fileBuf[recvBufferLength + j] = recvBuf[j]; } } file.write(fileBuf, 0, recvBufferLength); dos.writeUTF("1");//一个文件断点连续，以免超时 } else if (destBufferLength == 0||destBufferLength == -1) { System.out.println("传输文件:" + fileName + " 完成"); time = System.currentTimeMillis() - time; file.close(); break; } else { System.out.println("服务端拒绝传输"); file.close(); break; } } file.flush(); file.close(); } System.out.println("用时 : " + time + " 毫秒"); } } } catch (Exception e) { } finally { System.out.println("\n与服务器结束连接"); close(); } } public void downloadSmall() {// 下载小文件 try { FileOutputStream fos = null; String fileName = dis.readUTF();// 读取文件名称 String fileLength = dis.readUTF();// 读取文件长度 fos = new FileOutputStream(filePath + fileName); long time = System.currentTimeMillis(); System.out.println("\n接收文件:" + fileName + " ..."); int code = 0;// 临时存字节变量 while (true) { if ((code = is.read()) != 255) { fos.write(code); fos.flush(); } else { break; } } System.out.println("传输文件:" + fileName + " 完成"); System.out.println("用时：" + (System.currentTimeMillis() - time) + " 毫秒"); } catch (Exception e) { System.out.println(" 数据处理异常"); } } public void uploadSmall(File file2) {// 上传小文件 try { FileInputStream fis = null; fis = new FileInputStream(file2); int code = 0;// 临时存字节变量 dos.writeUTF(file2.getName());// 发送文件名称 dos.writeUTF(file2.length() + "");// 发送文件长度 System.out.println("\n上传文件:" + file2.getName() + " ..."); long time = System.currentTimeMillis(); while (true) { if ((code = fis.read()) != -1) {// 读到文件末尾 os.write(code);// 一个字节一个字节发送到客户端 } else { os.write(-1); break; } os.flush();// 清空缓存 } System.out.println("传输文件:" + file2.getName() + " 完成"); System.out.println("用时：" + (System.currentTimeMillis() - time) + " 毫秒"); fis.close(); } catch (Exception e) { System.out.println(" 数据处理异常"); } } public void close() {// 流及Socket关闭 try { if (null != is) is.close(); if (null != os) os.close(); if (null != dis) dis.close(); if (null != dos) dos.clos