多文件云传输系统框架

文章目录

多文件云传输系统框架
- 1. 需求分析
- 2.资源的表示
- - 2.1文件片段化处理
  - - 2.1.1文件片段头----- FileSectionHandle类
    - 2.1.2int与byte类型之间的转换----- TypeUtil类
    - 2.1.3 文件片段-----FileSection类
  - 2.2 资源基本信息-----ResourceBaseInfo类
  - 2.3 资源-----Resource类
  - 2.4 扫描本地资源 ----- Scanner类
- 3.系统结构及功能详解
- - 3.1 注册中心
  - 3.2 资源发送者
  - 3.3 资源请求者
- 4.分配策略
- - 4.1 资源分配策略
  - - 4.11 IResourceDistribution类
    - 4.12 ResourceDistributionStrategy
  - 4.2节点分配策略
  - - 4.2.1 INetNodeStrategy
    - 4.2.2 随机节点分配-----NetNodeStrategy类
    - 4.2.3 按节点的发送次数进行分配-----NetNodeSelectStrategy 类
- 5 文件指针池-----RandAccessFilePool类
- 6.断点续传的基础------UnReceiveSection类
- 7.注册中心代码
- - 7.1 注册中心启动类------RegisterCenter类
  - 7.2 注册中心RPC接口-----INodeAction
  - 7.3 注册中心RPC接口实现类 ----- NodeAction类
  - 7.4 ResourceName类
  - 7.5 ResourceNode类
  - 7.6 关于注册中心的侦听者模式-----ISpeaker、IListener
- 8.资源发送者
- 9 资源接受者

1. 需求分析

我们希望能实现如下功能：

当一个客户端请求资源时，会从注册中心得到拥有该资源的所有网络节点，该请求者会选出当前压力最小的K个发送者，并对他们请求这个资源的不同部分，最终请求者将会得到他请求的资源。
我们最终希望能实现断点续传。断点续传有两种情况：
情况1：比如发送端像请求端发送资源时，它还没发送完就下线了，最终接收端将接收不到完整的资源，我们希望接收端能清楚自己的哪些文件没有收到，并重新像其他拥有该资源的在线节点进行进行请求。
情况2： 接收端接收到一半，电脑关机了，我们希望他重新开机后，能从断点处重新下载，而不是重头下载。

2.资源的表示

我们把多个文件或者单个文件称作资源，比如QQ和微信就是两种资源。我们需要用一个类去描述该资源。

2.1文件片段化处理

文件传输需要用到网络，所以不可能将一个很大的文件一口气发过去，所以我们将一个文件片段话。
文件片段由两部分组成：文件片段头和该文件片段的内容。

2.1.1文件片段头----- FileSectionHandle类

/*** 文件片段头* 功能：* 1. 文件片段头的功能是为了描述一个文件片段属于哪一个文件，在此文件中的偏移量是多少，以及该片段内容的长度。* 2. 由于网络间的传输是以byte为单位的，所以我们需要提供将文件片段头变换为byte[]类型的方法，* 当然还需提供反变换的方法.* @author 田宜凡**/
public class FileSectionHandle {private int fileHandle; //文件片段所属文件的文件句柄，最终会通过这个文件句柄去找到该文件的相对路径，以及文件的大小。总体来说，文件句柄映射着一个文件。//为什么不直接把文件的路径直接替换掉fileHandle，因为你这是文件片段头，使用这些信息没用，而且文件路径的长度是不确定的。private int offset;  //该片段的偏移量private int len;//还片段的长度public FileSectionHandle() {}//三参构造函数public FileSectionHandle(int fileHandle, int offset, int len) {this.fileHandle = fileHandle;this.offset = offset;this.len = len;}//将三个int类型的成员变为byte类型并且放到同一个byte数组中public byte[] tobytes() {byte[] result = new byte[12];byte[] fileHandleBytes = TypeUtil.intToBytes(fileHandle);byte[] offsetBytes = TypeUtil.intToBytes(offset);byte[] lenBytes = TypeUtil.intToBytes(len);setBytes(result, 0, fileHandleBytes);setBytes(result, 0 + 4, offsetBytes);setBytes(result, 0 + 8, lenBytes);return result;}//将byte[]变换为真正的成员public FileSectionHandle(byte[] value) {byte[] fileHandleBytes = getByte(value, 0, 4); byte[] offsetBytes = getByte(value, 0 + 4, 8); byte[] lenBytes = getByte(value, 0 + 8, 12);fileHandle = TypeUtil.bytesToInt(fileHandleBytes);offset = TypeUtil.bytesToInt(offsetBytes);len = TypeUtil.bytesToInt(lenBytes);}void setBytes(byte[] resouce, int start, byte[] target) {int length = target.length;int end = start + length;for (int i = start ; i < end ; i++) {resouce[i] = target[i % length];}}byte[] getByte(byte[] resource, int start, int end) {int length = end - start;byte[] result = new byte[length];for (int i = 0 ; i < length ; i++) {result[i] = resource[i + start];}return result;}public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}public int getOffset() {return offset;}public void setOffset(int offset) {this.offset = offset;}public int getLen() {return len;}public void setLen(int len) {this.len = len;}@Overridepublic String toString() {return "fileHandle=" + fileHandle + ", offset=" + offset + ", len=" + len + "\n";}
}

2.1.2int与byte类型之间的转换----- TypeUtil类

/*** 功能；用于int类型与byte类型的转换(核心思想： 位运算)* @author ty**/
public class TypeUtil {public TypeUtil() {}/*** * @param 一个int的数据* @return 将int数据转换为byte[]类型*/public static byte[] intToBytes (int value) {byte[] result = new byte[4];for (int i = 0 ; i < 4 ; i++) {//如果将int强转为byte保留的是低八位result[i] = (byte) (value >> (8 * i));}return result;}/*** * @param 一个长度为4的byte[]数据* @return 将byte[]数据转换为int类型。*/public static int bytesToInt(byte[] value) {int length = value.length;int result = 0;for (int i = 0 ; i < length ; i++) {result |= ((((int)value[i]) & 0xFF) << (8 * i));}return result;}
}

2.1.3 文件片段-----FileSection类

/*** 用于表示一个文件片段，每一个文件片段都能通过文件句柄对应一个文件基本信息* @author 田宜凡**/
public class FileSection {//文件片段头private FileSectionHandle fileSectionHandle;//本片段的字节内容private byte[] value;public FileSection() {fileSectionHandle = new FileSectionHandle();}public FileSection(int fileHandle, int offset, int len) {this.fileSectionHandle = new FileSectionHandle();fileSectionHandle.setFileHandle(fileHandle);fileSectionHandle.setOffset(offset);fileSectionHandle.setLen(len);}public FileSectionHandle getFileSectionHandle() {return fileSectionHandle;}public void setFileSectionHandle(FileSectionHandle fileSectionHandle) {this.fileSectionHandle = fileSectionHandle;}public byte[] getValue() {return value;}public void setValue(byte[] value) {this.value = value;}public void setFileHandle(int fileHandle) {fileSectionHandle.setFileHandle(fileHandle);}public int getFileHandle() {return fileSectionHandle.getFileHandle();}public void setLen(int len) {fileSectionHandle.setLen(len);}public int getLen() {return fileSectionHandle.getLen();}public void setOffset(int offset) {fileSectionHandle.setOffset(offset);}public int getOffSet() {return fileSectionHandle.getOffset();}@Overridepublic String toString() {return fileSectionHandle.toString();}}

2.2 资源基本信息-----ResourceBaseInfo类

/*** 文件基本信息，对应着一个文件，文件片段会通过文件句柄，找到对应的文件基本信息* @author ty**/
public class ResourceBaseInfo {private int fileHandle;//文件句柄private String relativePath;//该文件的相对路径private long size;//该文件的大小public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}public String getRelativePath() {return relativePath;}public void setRelativePath(String relativePath) {this.relativePath = relativePath;}public long getSize() {return size;}public void setSize(long size) {this.size = size;}@Overridepublic String toString() {return "fileHandle=" + fileHandle + ", relativePath=" + relativePath + ", size=" + size + "\n";}
}

2.3 资源-----Resource类

package com.mec.ManyFile.resource;import java.util.List;
/*** 资源，一个资源可能是单文件也可能是多文件* @author ty**/
public class Resource {private String AppName;//资源名称private String absolutePath;//资源绝对根路径private String version;//资源的版本private List<FileSection> FileSectionList;//文件片段列表private List<ResourceBaseInfo> baseInfoList;//资源基本信息列表，请求时这个列表为空。public Resource() {    }public void setAbsolutePath(String absolutePath) {this.absolutePath = absolutePath;}public String getAppName() {return AppName;}public void setAppName(String appName) {AppName = appName;}public String getVersion() {return version;}public void setVersion(String version) {this.version = version;}public List<FileSection> getFileSectionList() {return FileSectionList;}public void setFileSectionList(List<FileSection> fileSectionList) {FileSectionList = fileSectionList;}public List<ResourceBaseInfo> getBaseInfoList() {return baseInfoList;}public void setBaseInfoList(List<ResourceBaseInfo> baseInfoList) {this.baseInfoList = baseInfoList;}public String getAbsolutePath() {return absolutePath;}//找到一个文件片段对对应的文件（资源基本信息）public ResourceBaseInfo getResourceBaseInfo(FileSection section) {int fileHandle = section.getFileHandle();return  getSameFileHandle(fileHandle);}//这个方法用于找到文件句柄相同的资源基本信息private ResourceBaseInfo getSameFileHandle(int filehandle) {for (ResourceBaseInfo rbi : baseInfoList) {int temp = rbi.getFileHandle();if (temp == filehandle) {return rbi;}}return null;}@Overridepublic String toString() {StringBuffer sb = new StringBuffer();sb.append(AppName + "\n"  + absolutePath + "\n"+ version+ "\n");if (FileSectionList != null) {for (FileSection filesection : FileSectionList) {sb.append(filesection.toString());}  }if (baseInfoList != null) {for (ResourceBaseInfo rbi : baseInfoList) {sb.append(rbi.toString());}}return sb.toString();}
}

2.4 扫描本地资源 ----- Scanner类

/*** 1.递归扫描本地文件* 2.自动获取对应文件的大小、相对路径* 3.自动添加文件句柄* @author ty**/
public class Scanner {public Scanner() {}/*** * @param appPath 资源所对应的根目录* @return*/public List<ResourceBaseInfo> ScannerAppPath(String appPath) {File file = new File(appPath);List<ResourceBaseInfo> rbiList = new ArrayList<>();explore(appPath, file, rbiList, 0);return rbiList;}int explore(String appPath, File file, List<ResourceBaseInfo> rbiList, int fileHandle) {File[] files = file.listFiles();for (File f : files) {if (f.isFile()) {fileHandle = creatResourceBaseInfo(appPath, f, rbiList, ++fileHandle);}if (f.isDirectory()) {fileHandle = explore(appPath, f,rbiList, fileHandle);}}return fileHandle;}int creatResourceBaseInfo(String appPath, File file, List<ResourceBaseInfo> rbiList, int fileHandle) {ResourceBaseInfo resourceBaseInfo = new ResourceBaseInfo();resourceBaseInfo.setFileHandle(fileHandle);resourceBaseInfo.setRelativePath(file.getAbsolutePath().replace(appPath + "\\", ""));resourceBaseInfo.setSize(file.length());rbiList.add(resourceBaseInfo);return fileHandle;}
}

3.系统结构及功能详解

3.1 注册中心

功能分析：

每当有客户端上线时，客户端都要以RPC方式连接注册测中心，以资源名称#资源版本号为键，比如 QQ#1，以NetNode类为值，向注册中心注册自己拥有的所有资源。
当一个发送者下线时，需要将自己的节点注销掉。正常下线还好办，但是异常下线我们就无法注销了，这显然是不合理的，当请求者RPC连接发送者时，如果连不上，会出现异常处理，这时再进行注销操作。
资源请求者可以从注册中心得到一个资源所对应网络节点列表。
注册中心可以动态更新每一个网络节点（NetNode）的发送次数。
除了注册资源名称还要注册每个资源的资源基本信息，比如该资源有多少个文件，每个文件有多大，以及文件的相对路径。这些都写在一个类里面。当然与之对应的也要有注销。得到资源基本信息的操作。

综上所述：注册中心主要有三个核心功能：注册、注销、得到列表。
NetNode类：
此类有三个成员

//网络节点
public class NetNode {private int port;//网络节点的端口号private String ip;//网络节点的IP地址private int sendingTime;该节点已经发送了多少次资源public NetNode(int port, String ip, int sendingTime) {super();this.port = port;this.ip = ip;this.sendingTime = sendingTime;}public NetNode() {}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public int getSendingTime() {return sendingTime;}public void setSendingCount(int sendingTime) {this.sendingTime = sendingTime;}public void increase() {sendingTime++;}public void crease() {sendingTime--;}public String toString() {return "NetNode [port=" + port + ", ip=" + ip + ", sendingTime=" + sendingTime + "]";}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + ((ip == null) ? 0 : ip.hashCode());result = prime * result + port;return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;NetNode other = (NetNode) obj;if (ip == null) {if (other.ip != null)return false;} else if (!ip.equals(other.ip))return false;if (port != other.port)return false;return true;}}

3.2 资源发送者

功能分析：

资源请求者会开启资源资源接收服务器，和RPC客户端。通过RPC的方式将自己资源接收服务器的ip地址和端口号以及要请求的资源基本信息发过去，让资源发送者连接资源请求者，并发送请求的资源。

3.3 资源请求者

功能分析：

为了实现断点续传，需要对接收的文件片端进行记录
每接收到一个文件片段。要根据接收端自己设定的绝对根路径，结合资源基本信息的中的相对路径把文件片段，写入到磁盘上。

4.分配策略

4.1 资源分配策略

4.11 IResourceDistribution类

/*** 不论是默认资源分配策略还是自定义资源分配策略都有应有* 1.默认文件片段大小* 2.最大文件片段大小* 3.分配单文件资源* 4.分配多文件资源* @author ty**/
public interface IResourceDistribution {long DEFAULT_SIZE = 1 << 14;long MAX_SIZE = 1 << 15;public List<List<FileSection>> divideResourceBaseInfos(ResourceBaseInfo rbi, List<NetNode> nodeList);public List<List<FileSection>> divideResourceBaseInfo(List<ResourceBaseInfo> rbis, List<NetNode> nodeList);
}

4.12 ResourceDistributionStrategy

/** 资源分配策略：* 我们收到了资源的信息列表，丛中我们可以知道* 1.每一个文件的句柄* 2.相对路径* 3.以及该文件的大小* 4.我们根据文件的大小进行分片，订一个默认的大小* 5.如果该文件的大小小于默认大小，不用进行分片，* 6.如果大于默认大小就要进行分片* 这个默认大小最终我们希望实现可配置。* */
public class ResourceDistributionStrategy implements IResourceDistribution{private long bufferSize = DEFAULT_SIZE;public ResourceDistributionStrategy() {}public void setBufferSize(long bufferSize) {if (bufferSize < 0 || bufferSize > MAX_SIZE) {return;}this.bufferSize = bufferSize;}//分配单文件资源public List<List<FileSection>> divideResourceBaseInfos(ResourceBaseInfo rbi, List<NetNode> nodeList) {List<ResourceBaseInfo> rbis = new ArrayList<ResourceBaseInfo>();rbis.add(rbi);List<List<FileSection>> result = divideResourceBaseInfo(rbis, nodeList);return result;}/*** 功能：<br>* 1.根据发送端列表得知发送端的个数* 2.遍历每个文件信息，对文件的大小进行分解* 3.最终得到和发送端个数一致的文件片段堆*     @param rbis 资源的所有文件列表*  @param nodeList 发送端列表*  @return 得到根据发送者的数量分配的文件片段列表*/public List<List<FileSection>> divideResourceBaseInfo(List<ResourceBaseInfo> rbis, List<NetNode> nodeList) {int sendCount = nodeList.size();int index = 0;List<List<FileSection>> result = new ArrayList<List<FileSection>>();for (int i = 0 ; i < sendCount ; i++) {List<FileSection> temp = new ArrayList<FileSection>();result.add(temp);}for (ResourceBaseInfo rbi : rbis) {long size = rbi.getSize();int fileHandle = rbi.getFileHandle();if (size < bufferSize) {FileSection fileSection = new FileSection(fileHandle, 0 ,(int)size);List<FileSection> secList = result.get(index);index = (index + 1) % sendCount;secList.add(fileSection);} else {long restSize =  size;int offset = 0;int len;while (restSize != 0) {len = (int) (restSize > bufferSize ? bufferSize : restSize);FileSection fileSection = new FileSection(fileHandle, offset ,(int)len);offset += len;restSize -= len;List<FileSection> secList = result.get(index);index = (index + 1) % sendCount;secList.add(fileSection);}}}return result;}}

4.2节点分配策略

关于节点分配我有两种想法，一种是随机分配，一种是根据每个节点已经发送的次数进行分配 INetNodeStrategy

4.2.1 INetNodeStrategy

/*** 接点分配策略接口* 1.设置默认发送次数* 2.设置最大发送次数* 3.选择网络节点* @author ty**/
public interface INetNodeStrategy {int DEFAULT_SENDER_COUNT = 3;int MAX_SENDER_COUNT = 20;List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists);
}

4.2.2 随机节点分配-----NetNodeStrategy类

/**
这里的节点分配采用的是随机的办法
1.根据需要发送端的个数，将整个接点列表分为多份
2.从每一份中随机挑选一个节点
@author ty* */
public class NetNodeStrategy implements INetNodeStrategy{private static int maxSendCount = DEFAULT_SENDER_COUNT;public NetNodeStrategy() {}public List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists) {int sendCount = netNodeLists.size();if (sendCount <= maxSendCount) {return netNodeLists;} else {return getSendNodeList(netNodeLists);}}private List<NetNode> getSendNodeList(List<NetNode> netNodeLists) {List<NetNode> netList = new ArrayList<NetNode>();int sendCount = netNodeLists.size();int oneGroupCount = sendCount / maxSendCount;int restCount = sendCount % maxSendCount;Random rand = new Random();for (int i = 0 ; i < maxSendCount ; i++) {int temp = i == (maxSendCount - 1) ? rand.nextInt(oneGroupCount + restCount): rand.nextInt(oneGroupCount);int index = temp + i * oneGroupCount;netList.add(netNodeLists.get(index));}return netList;}
}

4.2.3 按节点的发送次数进行分配-----NetNodeSelectStrategy 类

该分配策略的核心问题：从多个节点中找到发送次数的几个
一般我们会进行升序排序，然后选出最小的是几个，一般排序的时间复杂度为O(n^2),我采用的方法将时间复杂度控制到最大O(3n);
算法图解：

/** 这里的节点分配采用的找出发送次数最少的几个节点1.遍历节点列表2.找出发送次数最少的是三个节点* */
public class NetNodeSelectStrategy implements INetNodeStrategy{private int maxSenderCount = DEFAULT_SENDER_COUNT;public NetNodeSelectStrategy() {}public void setMaxSenderCount(int maxSenderCount) {this.maxSenderCount = maxSenderCount < MAX_SENDER_COUNT ? maxSenderCount : MAX_SENDER_COUNT;}@Overridepublic List<NetNode> SelectNetNdoe(List<NetNode> netNodeLists) {int sendCount = netNodeLists.size();if (sendCount <= maxSenderCount) {return netNodeLists;} else {return getMinSendNodeList(netNodeLists);}}//具体算法请看算法图解private List<NetNode> getMinSendNodeList(List<NetNode> netNodeLists) {List<NetNode> result = new ArrayList<NetNode>();NetNode maxNode = netNodeLists.get(0);for (int i = 1 ; i < netNodeLists.size() ; i++) {NetNode temp = netNodeLists.get(i);if (temp.getSendingTime() > maxNode.getSendingTime()) {maxNode = temp;}}int maxsendingTime = maxNode.getSendingTime();int[] sendCount = new int[maxsendingTime + 1];for(NetNode node : netNodeLists) {sendCount[node.getSendingTime()]++;}int maxSendCount = maxSenderCount;System.out.println("maxSendCount" +  maxSendCount);for (int i = 0 ; i < sendCount.length ; i++) {if (maxSendCount <= 0) {sendCount[i] = 0;continue;}maxSendCount -= sendCount[i]; System.out.println("s " +maxSendCount);if(maxSendCount < 0) {maxSendCount += sendCount[i];sendCount[i] = maxSendCount;}}for (NetNode netNode : netNodeLists) {int time = netNode.getSendingTime();if (sendCount[time] == 0) {continue;}sendCount[time]--;result.add(netNode);}return result;}
}

5 文件指针池-----RandAccessFilePool类

每一次给一个文件中的指定位置，写一个片段，都需要RandAccessFile对象，该对象用完后需要关闭,但是这个文件整体没有接受完的时候。
就会存在RandAccessFile对象不停的创建以及关闭的问题。这样很费时，所以以一个文件的路径为键，以文件指针为值将它缓存起来，只有当一个文件全部接收完时，我们在关闭它。

public class RandAccessFilePool {private Map<String, RandomAccessFile> rafPool;RandAccessFilePool() {rafPool = new ConcurrentHashMap<>();}RandomAccessFile getRaf(String filePath) {RandomAccessFile raf = rafPool.get(filePath);if (raf == null) {try {// TODO 根据filePath，创建相关目录raf = new RandomAccessFile(filePath, "rw");rafPool.put(filePath, raf);} catch (FileNotFoundException e) {e.printStackTrace();}}return raf;}void close(String filePath) {RandomAccessFile raf = rafPool.remove(filePath);raf.close();}
}

6.断点续传的基础------UnReceiveSection类

我们能实现断点续传的基础是UnReceiveSection类，而该类的基础是FileSection类，因为该类的对象保存了该文件片段在文件中的偏移量和长度。
我们使用一个FileSection的List来保存未接收的文件片段

/*** 这一个类对应一个文件，用来记录我们的文件有没有接受完毕* @author ty**/
public class UnReceiveSection {private int fileHandle//文件句柄private List<FileSection> unReceiveList;//未接收文件片段列表//初始化时，给unReceiveList加入一个文件片段，偏移量为0，片段长度为文件长度public UnReceiveSection(int fileHandle, int fileSize) {unReceiveList = new LinkedList<FileSection>();this.fileHandle = fileHandle;FileSection fileSection = new FileSection();fileSection.setFileHandle(fileHandle);fileSection.setLen(fileSize);fileSection.setOffset(0);unReceiveList.add(fileSection);}public int getFileHandle() {return fileHandle;}public void setFileHandle(int fileHandle) {this.fileHandle = fileHandle;}//给unReceiveList加入一个文件片段，具体思想请看上图public void addFileSection(FileSection fileSection) {int fileHandle = fileSection.getFileHandle();if (fileHandle != this.fileHandle) {return;}int targetOffset = fileSection.getOffSet();int targetLen = fileSection.getLen();int targetIndex = targetLen + targetOffset;FileSection section = getDivideSection(targetOffset, targetLen);if (section == null) {return;}int srcOffset = section.getOffSet();int srcLen = section.getLen();int srcIndex = srcOffset + srcLen;if (targetOffset == srcOffset&& targetIndex == srcIndex) {return;}int leftOffset = srcIndex;int leftLen = targetOffset - srcOffset;int rightOffset = targetOffset + targetOffset;int rightLen = srcOffset + srcLen - rightOffset;if(leftLen != 0) {unReceiveList.add(new FileSection(fileHandle, leftOffset, leftLen));}if(rightLen != 0) {unReceiveList.add(new FileSection(fileHandle, rightOffset, rightLen));}}private FileSection getDivideSection(int offset, int len) {int targetIndex = offset + len;FileSection fileSection = null;for (FileSection section : unReceiveList) {int tmpOffset = section.getOffSet();int tmpLen = section.getLen();int srcIndex = tmpOffset + tmpLen;if (offset >= tmpOffset && targetIndex <= srcIndex) {fileSection = section;}}unReceiveList.remove(fileSection);return fileSection;}//unReceiveList为空代表列表该文件接收完成public boolean isFinish() {if (unReceiveList.size() == 0) {return true;}return false;}public List<FileSection> getUnReceiveFileSection() {return unReceiveList;}
}

7.注册中心代码

当初有想过资源发送端与注册中心之间进行长连接，因为异常掉线后，注册中心可以及时注销掉该节点，预防资源求者得到已经下线的节点列表，但是不论是资源请求端还是资源发送端。对于App服务器来说都为客户端，所以这个数量很大，如果与注册中心长连接的话，注册中心的压力很大，为了缓解这种压力，我们采用短连接（RPC）。
短连接解决资源发送端异常掉线问题

短连接就是所谓的一回合，无状态连接，所以注册中心无法得知异常掉线
虽然可以在资源请求端RPC资源发送端时的以异常里进行注销的操作，然后采用其他可用的节点。这样就会产生一个问题：请求得到节点列表中掺杂了很多用不了的节点。
所以我们打算比如每半天对注册中心进行一次心跳，让它短连接所有的资源发送者，成功什么都不做，连接失败了就注销该节点，这样可以定期销毁不能使用的网络节点。但是我尝试了很多方式，都行不通。起初我本来想使用每个资源发送者的RPC服务器。我可以让注册中心连接资源发送者的RPC服务器进行一次短连接，如果连接失败，注册中心就可以认定资源发送者掉线了，听起来很美好，但是判断连接失败的时间太长了。长到系统根本无法接受，所以这也是我这个系统留下的最大的遗憾。

7.1 注册中心启动类------RegisterCenter类

/*** 注册中心功能：* 1.开启RPC服务器服务器* 2.正常关闭RPC服务器* 3.提供默认端口号* @author ty**/
public class RegisterCenter implements ISpeaker{private  RMIServer rmiServer;private int rmiPort;private static final int RMIDEFAULT_PORT = 54199; private List<IListener> listenrList;public RegisterCenter() {this(RMIDEFAULT_PORT);}public RegisterCenter(int rmiPort) {this.rmiPort = rmiPort;}public void setListenrList(List<IListener> listenrList) {this.listenrList = listenrList;}public void setRmiPort(int rmiPort) {this.rmiPort = rmiPort;}public void startup() {reportMessage("正在开启注册中心.....");rmiServer = new RMIServer(rmiPort);reportMessage("注册中心开启成功.....");rmiServer.startup();reportMessage("短连接服务器开始侦听客户端");rmiServer.registory("com.mec.ManyFile.RegistCenter");}public void shutdown() {rmiServer.close();reportMessage("短连接服务器正常关闭...");}@Overridepublic void addListener(IListener iListtener) {if (listenrList == null) {listenrList  = new ArrayList<>(); }if (listenrList.contains(iListtener)) {return; }listenrList.add(iListtener);}@Overridepublic void removeListener(IListener iListtener) {if (!listenrList.contains(iListtener)) {return; }listenrList.remove(iListtener);}public void reportMessage(String message) {if (listenrList == null || listenrList.size() == 0) {return;}for (IListener listen : listenrList) {listen.dealMessage(message);}}

7.2 注册中心RPC接口-----INodeAction

/*** 此接口更包含了注册中心所拥有的功能，为了RPC的调用* @author ty**/
public interface INodeAction {void logoutNode(NetNode node);//注销一个节点void registerNode(ResourceName service, NetNode node);//注册一个节点List<NetNode> getNodeList(ResourceName service);//得到一个资源的节点列表void logout(ResourceName res);//注销一个资源的信息void register(ResourceName service, Resource res);//注册一个资源的信息Resource getResource(ResourceName service);//得到资源信息void inCreaseSendCount(NetNode node);//增加一个节点的发送次数void CreaseSendCount(NetNode node);//减少一个节点的发送次数
}
}

7.3 注册中心RPC接口实现类 ----- NodeAction类

我认为此类中的注销节点，增加发送次数，减少发送次数这三个操作写的时间复杂度都很高，但是我们希望这里能快速执行完，所有我觉得这里处理不是很好，希望以后会有更好的办法。

/*** 注册中心RPC实现类* @author ty**/
@Interfaces(interfacees = {INodeAction.class})
public class NodeAction implements INodeAction{/*** relationMap * 键为 资源名#版本号字符串  * 值为 该资源的资源信息以及拥有该资源的网络节点*/private static Map<String, ResourceNode> relationMap = new ConcurrentHashMap<String, ResourceNode>();//遍历每一个键所对应的节点列表，并在每一个节点列表中找到要注销的NetNode，然后删除//这种方法我感觉时间复杂度很高，不是很满意，希望以后可以改进@Overridepublic void logoutNode(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {temp = one;break;}  }nodeList.remove(temp);}}/***     注册一个节点* 根据键值在relationMap中找到有没有对应的值，没有的话初始化一个值放进去，再把node放进去*/@Overridepublic void registerNode(ResourceName service, NetNode node) {String key = service.toString();ResourceNode resNode = relationMap.get(key);if(resNode == null) {resNode = new ResourceNode();relationMap.put(key, resNode);}List<NetNode> NodeList = resNode.getNetNodes();if (NodeList.contains(node)) {return;}NodeList.add(node);}/*** 得到节点列表* 1.首先得判断你寻求的节点信息列表存不存在* 2.存在的话返回，不存在返回null*/@Overridepublic List<NetNode> getNodeList(ResourceName service) {ResourceNode node = relationMap.get(service.toString());if (node == null) {return null;}return node.getNetNodes();}/*** 注销资源信息，这是由APP服务器做的事情* 如果资源信息都了被服务器删除了，关于这个资源的节点列表也要删除*因为资源已经被APP服务器抛弃了*/@Overridepublic void logout(ResourceName service) {String key = service.toString();if(relationMap.get(key) == null) {return;}relationMap.remove(key);}/*** 注册资源信息，由APP服务器进行* 先判断在HashMap中键存不存在。没有的话要先初始化*/@Overridepublic void register(ResourceName service, Resource res) {ResourceNode resNode = relationMap.get(service);if(resNode == null) {resNode = new ResourceNode();relationMap.put(service.toString(), resNode);}resNode.setRes(res);}/*** 得到资源信息* 从relationMap中根据传进来的键得到资源信息* 如找找不到，返回null*/@Overridepublic Resource getResource(ResourceName service) {ResourceNode node = relationMap.get(service.toString());if (node == null) {return null;}return node.getRes();}/**增加发送次数，节点分配策略就是基于此数值的，所以每当一个发送端被分配出去，就我们就要通过RPC使这个节点的发送次数次数加一**/@Overridepublic void inCreaseSendCount(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {one.increase();break;}  }}}/**减少发送次数，节点分配策略就是基于此数值的，每当一个节点发送完毕，既然让个值减一**/@Overridepublic void CreaseSendCount(NetNode node) {Set<String> keyset = relationMap.keySet();Iterator<String> set = keyset.iterator();while(set.hasNext()) {String key = set.next();ResourceNode resNode = relationMap.get(key);List<NetNode> nodeList = resNode.getNetNodes();NetNode temp = null;for (NetNode one : nodeList) {if (one.equals(node)) {one.crease();break;}    }}}
}

7.4 ResourceName类

/*** 注册中心的关系表中的键* 1.资源的名称* 2.资源的版本* 最终根据toString()方法，以字符串的身份作为键
**/
public class ResourceName {String appName;String version;public ResourceName() {}public ResourceName(ResourceName resourceName) {this.appName = resourceName.getAppName();this.version = resourceName.getVersion();}public String getAppName() {return appName;}public void setAppName(String appName) {this.appName = appName;}public String getVersion() {return version;}public void setVersion(String version) {this.version = version;}@Overridepublic String toString() {return appName + "#" + version;}}

7.5 ResourceNode类

/*** 注册中心关系表中的值，由两部分组成* 1.资源信息* 2.节点信息列表* @author ty**/
public class ResourceNode{private Resource res;private List<NetNode> netNodes;public ResourceNode() {netNodes = new LinkedList<>();}public Resource getRes() {return res;}public void setRes(Resource res) {this.res = res;}public List<NetNode> getNetNodes() {return netNodes;}public void setNetNode(List<NetNode> netNode) {this.netNodes = netNode;}
}

7.6 关于注册中心的侦听者模式-----ISpeaker、IListener

侦听者机制的作用
比如服务器器开启后你希望向界面上输出一些东西，或者将一些信息写入日志，但是当前状态下并没有界面，只有信息，如何把这个信息传递到未来才可能出现的界面上，侦听者机制就可以很好的处理这个问题，首先侦听者机制有两个重要的接口ISpeaker、IListener

public interface ISpeaker {void addListener(IListener iListtener);void removeListener(IListener iListtener);
}

public interface IListener {void dealMessage(String message);
}

注册中心实现ISpeaker接口，未来的界面实现IListener接口

//此段代码截取了RegisterCenter类的一部分内容
//这些内容就是实现侦听者模式的全部，并不负载，但是要求你要对接口很熟悉
private List<IListener> listenrList;@Overridepublic void addListener(IListener iListtener) {if (listenrList == null) {listenrList  = new ArrayList<>(); }if (listenrList.contains(iListtener)) {return; }listenrList.add(iListtener);}@Overridepublic void removeListener(IListener iListtener) {if (!listenrList.contains(iListtener)) {return; }listenrList.remove(iListtener);}public void reportMessage(String message) {if (listenrList == null || listenrList.size() == 0) {return;}for (IListener listen : listenrList) {listen.dealMessage(message);}}

在注册中心使用时，你只需要调用reportMessage(String message)方法把你要传递出去的信息作为参数传进去就行。具体把信息输出到哪里，还得看IListenner的实现类怎么去写dealMessage(String message)方法，最后再将IListener的实现类通过void addListener(IListener iListtener)方法提前加进去即可。

8.资源发送者

 /*发送者：* 1.收到对方发来的资源请求，以及请求者的ip 和 port* 2.连接接收者服务器* 3.从本地中提取文件片段* 4.提取一个发送一个* */
public class Send {private Socket socket;RMIServer rmiServer;private String ip;private int port;private int RMIport;private DataOutputStream dos;private RafPool rafPool;public Send() { this("192.168.181.1",54188);}public Send(String ip, int port) {this.ip = ip;this.port = port;rafPool = new RafPool();}//初始化RMI服务器public void initRMIServer() {rmiServer = new RMIServer(RMIport);rmiServer.startup();rmiServer.registory("com.mec.ManyFile.send");}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public int getRMIport() {return RMIport;}public void setRMIport(int rMIport) {RMIport = rMIport;}public void connectToServer() {try {socket = new Socket(ip, port);dos = new DataOutputStream(socket.getOutputStream()); } catch (UnknownHostException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();} }void sendFileSection(FileSection fileSection) {FileSectionHandle fileHandle = fileSection.getFileSectionHandle();byte[] fileHandleByte = fileHandle.tobytes();byte[] value = fileSection.getValue();try {dos.write(fileHandleByte, 0, 12);dos.write(value,0, fileHandle.getLen());} catch (IOException e) {e.printStackTrace();}}//从本地中读取这个文件片段public FileSection getFileSectionFromNative(FileSection section, String filePath) {RandomAccessFile raf = rafPool.get(filePath);int offset = section.getOffSet();int len = section.getLen();byte[] result = null;try {raf.seek(offset);result = new byte[len];raf.read(result);section.setValue(result);} catch (IOException e) {e.printStackTrace();}return section;}//这个参数resource拥有所有信息public void sendResource(Resource resource) {String absoluPath = resource.getAbsolutePath();List<FileSection> sectionList = resource.getFileSectionList();for (FileSection section : sectionList) {ResourceBaseInfo rbi = resource.getResourceBaseInfo(section);//通过这个rbi和section就可以得到这个文件片段的路径String relaPath = rbi.getRelativePath();String filePath = absoluPath + "\\" + relaPath;FileSection resultSection = getFileSectionFromNative(section, filePath);sendFileSection(resultSection);}}public void close() {if (socket != null) {try {socket.close();} catch (IOException e) {} finally {socket = null;}}if (dos != null) {try {dos.close();} catch (IOException e) {} finally {dos = null;}}}public void RMIServerClose() {rmiServer.close();}
}

//资源请求接口
public interface IResquset {void send(NetNode receiver, Resource res);
}

/*** 连接资源接收者服务器* 发送资源* @author ty**/
public class Resquest implements IResquset{public Resquest() {}@Overridepublic void send(NetNode receiver, Resource res) {String ip = receiver.getIp();int port = receiver.getPort();Send send = new Send(ip, port);send.connectToServer();ResourceInfoPool infoPool = new ResourceInfoPool();ResourceName name = new ResourceName();name.setAppName(res.getAppName());name.setVersion(res.getVersion());Resource src = infoPool.gets(name);res.setBaseInfoList(src.getBaseInfoList());send.sendResource(res);}
}

9 资源接受者

1.资源接收者控制类

package com.mec.ManyFile.receive;import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;import com.mec.ManyFile.resource.FileSection;
import com.mec.ManyFile.resource.Resource;
import com.mec.ManyFile.resource.ResourceBaseInfo;
import com.mec.ManyFile.resource.UnReceiveSection;
import com.mec.rmi.core.IRMIAction;
import com.mec.rmi.core.RMIAction;
import com.mec.rmi.core.RMIClient;
/*** 资源接收者服务器* 1.开启等待资源发送者的连接* 2.每等待一个开启一个线程去完成接受* @author ty**/
public class Receive implements Runnable{private ServerSocket receiveServer;private int port;private volatile boolean goon; private Resource resource; //保存每个文件的未接收片段private Map<Integer, UnReceiveSection> unReceiveMap; private IRMIAction iConnectError;public Receive() { this(54188);}public Receive(int port) {this.port = port;this.unReceiveMap = new ConcurrentHashMap<Integer, UnReceiveSection>();iConnectError = new RMIAction();}public void setPort(int port) {this.port = port;}public Resource getResource() {return resource;}public void setiConnectError(IRMIAction iConnectError) {this.iConnectError = iConnectError;}public void setResource(Resource resource) {this.resource = resource;}public void startUp() {if (port == 0) {return;}if (goon == true) {return;}try {receiveServer = new ServerSocket(port);goon = true;new Thread(this).start();initUnReceiveMap();} catch (IOException e) {e.printStackTrace();}}//初始化每一个文件对应的未接受片段列表private void initUnReceiveMap() {List<ResourceBaseInfo> resList = resource.getBaseInfoList();for (ResourceBaseInfo res : resList) {int fileSize = (int) res.getSize();int fileHandle = res.getFileHandle();UnReceiveSection unReceiveSection = new UnReceiveSection(fileHandle, fileSize);unReceiveMap.put(fileHandle, unReceiveSection);}}public void shutdown() {close();}public <T> T getProxy(String ip, int port, Class<?> clazz) {RMIClient rmiClient = new RMIClient();rmiClient.setIp(ip);rmiClient.setPort(port);rmiClient.setRmiAction(iConnectError);return rmiClient.getProxy(clazz);}private void close() {if(goon == false) {return;}try {receiveServer.close();} catch (IOException e) {e.printStackTrace();} finally {goon = false;}}//整合每个文件的未接收的文件片段，将它们整合到一个列表里public List<FileSection> getUnFileSection() {List<FileSection> sectionList = new ArrayList<FileSection>();Set<Integer> keys = unReceiveMap.keySet();for (Integer key : keys) {UnReceiveSection unRec = unReceiveMap.get(key);if (!unRec.isFinish()) {sectionList.addAll(unRec.getUnReceiveFileSection());}}return sectionList;}@Overridepublic void run() {while (goon) {try {Socket socket = receiveServer.accept();new DealReceive(socket, resource, unReceiveMap);} catch (IOException e) {//文件接收服务器异常掉线}}}}

2.处理接受的资源

package com.mec.ManyFile.receive;import java.io.DataInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.Socket;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;import com.mec.ManyFile.resource.FileSectionHandle;
import com.mec.ManyFile.resource.Resource;
import com.mec.ManyFile.resource.ResourceBaseInfo;
import com.mec.ManyFile.resource.UnReceiveSection;
import com.mec.ManyFile.Core.RafPool;
import com.mec.ManyFile.resource.FileSection;/*** 处理每个发送端发来的文件片段* @author ty**/
public class DealReceive implements Runnable{private Socket socket;private DataInputStream dis;private static final int BUFFER_SIZE = 1 << 10;private boolean goon;private Resource resource;private List<FileSection> fileSectionPool;private Map<Integer, UnReceiveSection> unReceiveMap;DealReceive(Socket socket, Resource resource, Map<Integer, UnReceiveSection> unReceiveMap) {fileSectionPool = new LinkedList<FileSection>();this.resource = resource;this.socket = socket;this.unReceiveMap = unReceiveMap;try {dis = new DataInputStream(socket.getInputStream());} catch (IOException e) {e.printStackTrace();} new Thread(this).start();new Thread(new DealFileSection()).start();goon = true;}/*** 读取len个字节，我们的缓冲区不一定能快速的收纳len个字节* 采用以下方式能准确的读取字节流* @param size* @return*/byte[] readBytes(int size) {int restLen = size;int readLen = 0;int len = size;int offset = 0;byte[] result = new byte[restLen];while(restLen > 0) {len = restLen < BUFFER_SIZE ? restLen : BUFFER_SIZE;try {readLen = dis.read(result, offset, len);restLen -= readLen;offset += readLen;} catch (IOException e) {goon = false;close();}}return result;}/*** 读取一个文件片段* @return*/FileSection readFileSection() {FileSection fileSection = new FileSection();byte[] fileHand = readBytes(12);FileSectionHandle fileHandle = new FileSectionHandle(fileHand); byte[] value = readBytes(fileHandle.getLen());fileSection.setFileSectionHandle(fileHandle);fileSection.setValue(value);return fileSection;}void close() {if (dis != null) {try {dis.close();} catch (IOException e) {} finally {dis = null;}}if (socket != null) {try {socket.close();} catch (IOException e) {} finally {socket = null;}}}@Overridepublic void run() {while(goon) {FileSection fileSection = readFileSection();//每读取一个，把他放入到缓冲区里，提高读取效率，用另一个线程完成本地的写fileSectionPool.add(fileSection);//dealFileSection(fileSection);}}/*** 将缓冲区中的文件片段根据资源信息慢慢的写入到本地去* @author ty**/class DealFileSection implements Runnable{DealFileSection() {}@Overridepublic void run() {String absolutePath = resource.getAbsolutePath();RafPool rafPool = new RafPool();while(goon || !fileSectionPool.isEmpty()) {if (fileSectionPool.isEmpty()) {continue;}FileSection fileSection = fileSectionPool.remove(0);ResourceBaseInfo rbi = resource.getResourceBaseInfo(fileSection);String relativePath = rbi.getRelativePath();String filePath = absolutePath + "\\" + relativePath;RandomAccessFile raf = rafPool.get(filePath);readFileFromNative(raf, fileSection);}rafPool.closeAll();}private boolean readFileFromNative(RandomAccessFile raf, FileSection fileSection) {int fileHandle = fileSection.getFileHandle();int offset = fileSection.getOffSet();byte[] result = fileSection.getValue();try {raf.seek(offset);raf.write(result);UnReceiveSection unReceiveSection = unReceiveMap.get(fileHandle);unReceiveSection.addFileSection(fileSection);} catch (IOException e) {e.printStackTrace();}return true;}}}

总体来说，这写内容有很多瑕疵，不过这已经榨干本博主能力的极限了，希望以后能在这条路上越走越远！！！！